1#![forbid(unsafe_code)]
2
3use std::{
4 path::{Path, PathBuf},
5 sync::Arc,
6};
7
8use anyhow::Context;
9use kcode_kennedy_cli::{Args, Command};
10use kcode_speaker_system::SpeechClassifier;
11
12const SPEECH_CLASSIFICATION_DATABASE_PATH: &str = "./data/kennedy-speech-classification.sqlite3";
13
14#[tokio::main]
15pub async fn main() -> anyhow::Result<()> {
16 tracing_subscriber::fmt()
17 .with_env_filter(
18 tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| {
19 "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()
20 }),
21 )
22 .init();
23 rustls::crypto::ring::default_provider()
24 .install_default()
25 .map_err(|_| anyhow::anyhow!("installing TLS crypto provider"))?;
26 let mut args = kcode_kennedy_cli::parse();
27 let vault_path = args.vault_path.clone();
28 match args.command.take() {
29 Some(Command::Secrets { command }) => {
30 let _maintenance_guard = tokio::net::TcpListener::bind(&args.kweb_bind)
31 .await
32 .with_context(|| {
33 format!(
34 "binding maintenance lock {}; stop the running Kennedy server before changing its credential vault",
35 args.kweb_bind
36 )
37 })?;
38 kcode_kennedy_bootstrap_secrets::manage(command, &vault_path)
39 }
40 Some(Command::KmapSize) => {
41 let _maintenance_guard =
42 maintenance_guard(&args.kweb_bind, "measuring the Kweb").await?;
43 let kweb_config = kcode_kennedy_bootstrap_secrets::unlock_kweb(&vault_path)?;
44 let size = kcode_kmap_size::measure(&args.kweb_root, kweb_config)?;
45 println!("{}", kcode_kmap_size::render(&size));
46 Ok(())
47 }
48 None => run_server(args, vault_path).await,
49 }
50}
51
52async fn run_server(args: Args, vault_path: PathBuf) -> anyhow::Result<()> {
53 let kweb_listener = tokio::net::TcpListener::bind(&args.kweb_bind)
54 .await
55 .with_context(|| format!("binding Kweb listener {}", args.kweb_bind))?;
56 ensure_runtime_parent_directories(&args, &vault_path)?;
57
58 let kcode_kennedy_bootstrap_secrets::ServerSecrets {
59 openai_api_key,
60 gemini_api_key,
61 telegram_bot_token,
62 crates_io_key,
63 kweb_config,
64 } = kcode_kennedy_bootstrap_secrets::unlock_server(&vault_path)?;
65 let telegram_bot_token = telegram_bot_token
66 .map(kcode_tg_kennedy_bot::BotToken::new)
67 .transpose()?;
68
69 let codex_catalog_cache =
70 kcode_codex_runtime::CatalogCache::new(kcode_codex_runtime::DEFAULT_CODEX_EXECUTABLE);
71 let (kmap, system_roots) =
72 kcode_kennedy_roots::open(&args.kweb_root, kweb_config, &args.user_database)?;
73 let (kmap_commands, kmap_command_runtime) =
74 kcode_kmap_command_lane::open(&args.user_database, kmap.clone())?;
75 let credits = kcode_credits::Credits::open(&args.credits_database)?;
76 let task_board = kcode_task_board::TaskBoard::open(&args.task_board_database, credits.clone())?;
77 let speech_classifier = SpeechClassifier::open(SPEECH_CLASSIFICATION_DATABASE_PATH)
78 .with_context(|| {
79 format!("opening speaker-classification database {SPEECH_CLASSIFICATION_DATABASE_PATH}")
80 })?;
81 let speech_classifier = Arc::new(speech_classifier);
82 let dev_tools = kcode_dev_tools::Service::open(kcode_dev_tools::Config {
83 rust_libraries_root: args.rust_libs_root.clone(),
84 web_libraries_root: args.web_libs_root.clone(),
85 web_publications_root: args.web_libs_published_root.clone(),
86 rust_binaries_root: args.rust_bins_root.clone(),
87 rust_binary_publications_root: args.rust_bin_artifacts_root.clone(),
88 crates_io_registry_token: crates_io_key,
89 })
90 .map_err(anyhow::Error::new)
91 .with_context(|| {
92 format!(
93 "opening managed Kcode development roots under {}",
94 args.rust_libs_root
95 .parent()
96 .unwrap_or(Path::new("."))
97 .display()
98 )
99 })?;
100 let web_publications_root = dev_tools.web_publications_root().to_path_buf();
101 let telegram_identity = std::sync::Arc::new(kcode_telegram_identity::Directory::open(
102 &args.user_database,
103 &args.telegram_bootstrap_username,
104 )?);
105 let history_service =
106 kcode_session_history::SessionHistory::open(kcode_session_history::Config {
107 directory: args.session_directory,
108 completed_list: args.session_history_file,
109 provider_cost_compatibility: Some(
110 kcode_intelligence_chatend::provider_cost_compatibility(),
111 ),
112 })?;
113 let (intelligence_service, intelligence_runtime) =
114 kcode_intelligence_router::open(kcode_intelligence_router::Config {
115 openai_api_key,
116 gemini_api_key,
117 codex_catalog_cache,
118 receipt_directory: args.intelligence_usage_directory,
119 })
120 .await?;
121 let agent_runtime = kcode_agent_runtime::AgentRuntime::new(intelligence_service.clone());
122 let telegram_runtime = kcode_tg_kennedy_bot::open(kcode_tg_kennedy_bot::Config {
123 database: args.telegram_database,
124 bot_token: telegram_bot_token,
125 identity_sink: telegram_identity.clone(),
126 max_voice_bytes: args.telegram_max_voice_bytes,
127 })
128 .await?;
129 let telegram_service = telegram_runtime.service();
130
131 let chunk_intelligence = intelligence_service.clone();
132 let transcribe_chunk: kcode_audio_ingress::AudioChunkCall = Arc::new(move |request| {
133 let intelligence = chunk_intelligence.clone();
134 Box::pin(async move {
135 let user = intelligence
136 .for_user(request.user_id)
137 .map_err(audio_intelligence_error)?;
138 let media = kcode_intelligence_router::Media::audio(
139 request.audio_ogg,
140 "audio-chunk.ogg",
141 "audio/ogg",
142 )
143 .map_err(audio_intelligence_error)?;
144 user.analyze_audio(kcode_intelligence_router::AudioAnalysisRequest {
145 operation: "transcribe_chunk".into(),
146 prompt: request.prompt,
147 model: request.model,
148 media,
149 schema: request.schema,
150 max_output_tokens: request.max_output_tokens,
151 temperature: None,
152 operation_id: uuid::Uuid::new_v4(),
153 parent_operation_id: None,
154 })
155 .await
156 .map(|response| response.value.text)
157 .map_err(audio_intelligence_error)
158 })
159 });
160 let text_intelligence = intelligence_service.clone();
161 let generate_text: kcode_audio_ingress::TextGenerationCall = Arc::new(move |request| {
162 let intelligence = text_intelligence.clone();
163 Box::pin(async move {
164 let reasoning_effort = match request.reasoning_effort.as_str() {
165 "xhigh" => kcode_intelligence_router::ReasoningEffort::XHigh,
166 _ => {
167 return Err(kcode_audio_ingress::IntelligenceError::new(
168 "AudioIngress requested an unsupported reasoning effort.",
169 false,
170 ));
171 }
172 };
173 let user = intelligence
174 .for_user(request.user_id)
175 .map_err(audio_intelligence_error)?;
176 user.generate_text(kcode_intelligence_router::TextGenerationRequest {
177 operation: request.operation,
178 prompt: request.prompt,
179 model: request.model,
180 reasoning_effort,
181 timeout: request.timeout,
182 operation_id: uuid::Uuid::new_v4(),
183 parent_operation_id: None,
184 })
185 .await
186 .map(|response| response.value.text)
187 .map_err(audio_intelligence_error)
188 })
189 });
190 let audio_transcriber =
191 kcode_audio_ingress::AudioTranscriber::new(transcribe_chunk, generate_text);
192 let audio = kcode_audio_ingress::AudioIngress::open(
193 &args.audio_ingress_directory,
194 audio_transcriber,
195 Arc::clone(&speech_classifier),
196 )
197 .await?;
198 let audio_coordinator = kcode_audio_session_ingress::Coordinator::new(
199 audio,
200 history_service.clone(),
201 kcode_audio_session_ingress::Config {
202 user_id: system_roots.user.to_string(),
203 effective_context_tokens: intelligence_runtime.context_window_tokens,
204 },
205 )?;
206 let http_router = kcode_http_api::router(kcode_http_api::Config {
207 kmap: kmap.clone(),
208 kmap_commands,
209 user_root_node_id: system_roots.user,
210 kennedy_root_node_id: system_roots.kennedy,
211 telegram: telegram_service.clone(),
212 session_history: history_service.clone(),
213 audio_ingress: audio_coordinator.clone(),
214 audio_max_upload_bytes: args.audio_ingress_max_upload_bytes,
215 task_board: task_board.clone(),
216 credits,
217 web_publications_root,
218 })?;
219 let orchestration_config = kcode_kennedy_orchestration::Config {
220 user_root_node_id: system_roots.user.to_string(),
221 kennedy_root_node_id: system_roots.kennedy.to_string(),
222 telegram_max_media_bytes: args.telegram_max_voice_bytes,
223 runtime_model: kcode_kennedy_orchestration::RuntimeModel::from_intelligence(
224 intelligence_runtime,
225 ),
226 };
227 let telegram_sessions = kcode_telegram_session_coordinator::Service::new(
228 telegram_service.clone(),
229 telegram_identity.clone(),
230 );
231 let session_service =
232 kcode_kennedy_sessions::Service::new(kcode_kennedy_sessions::Capabilities {
233 load_fixed_connections: args.fixed,
234 kmap: kmap.clone(),
235 intelligence: intelligence_service.clone(),
236 agents: agent_runtime,
237 history: history_service.clone(),
238 speech_classifier,
239 dev_tools: dev_tools.clone(),
240 telegram: telegram_sessions,
241 })
242 .with_task_board(task_board);
243 let orchestration_api = kcode_kennedy_orchestration::Api::new(
244 &orchestration_config,
245 kcode_kennedy_orchestration::LocalServices {
246 kmap: kmap.clone(),
247 intelligence: intelligence_service,
248 history: history_service.clone(),
249 audio: audio_coordinator,
250 directory: telegram_identity.clone(),
251 dev_tools,
252 telegram: telegram_service,
253 },
254 );
255 let orchestration_worker = kcode_kennedy_orchestration::build(
256 orchestration_config,
257 orchestration_api,
258 session_service,
259 );
260 let directory_roots = kcode_kennedy_roots::DirectoryRoots::new(
261 kmap,
262 telegram_identity,
263 args.telegram_bootstrap_username.clone(),
264 system_roots.user,
265 orchestration_worker.writer().clone(),
266 );
267 let telegram_session_runtime = Arc::new(kcode_kennedy_telegram_runtime::Runtime::new(
268 kcode_kennedy_telegram_runtime::Config {
269 telegram_max_media_bytes: args.telegram_max_voice_bytes,
270 telegram_web_user_handle: args.telegram_bootstrap_username,
271 },
272 orchestration_worker.clone(),
273 directory_roots,
274 ));
275 tokio::try_join!(
276 async {
277 kcode_http_api::serve(kweb_listener, http_router)
278 .await
279 .map_err(anyhow::Error::new)
280 },
281 telegram_runtime.run(),
282 kcode_kennedy_orchestration::run(orchestration_worker),
283 telegram_session_runtime.run(),
284 async { kmap_command_runtime.await.map_err(anyhow::Error::new) },
285 )?;
286 Ok(())
287}
288
289fn audio_intelligence_error(
290 error: kcode_intelligence_router::Error,
291) -> kcode_audio_ingress::IntelligenceError {
292 let retryable = error.retryable();
293 kcode_audio_ingress::IntelligenceError::new(error.message(), retryable)
294}
295
296fn ensure_runtime_parent_directories(args: &Args, vault_path: &Path) -> anyhow::Result<()> {
297 for path in [
298 vault_path,
299 &args.kweb_root,
300 &args.conversation_history_database,
301 &args.session_directory,
302 &args.session_history_file,
303 &args.telegram_database,
304 &args.user_database,
305 &args.task_board_database,
306 &args.credits_database,
307 Path::new(SPEECH_CLASSIFICATION_DATABASE_PATH),
308 &args.audio_ingress_directory,
309 &args.intelligence_usage_directory,
310 &args.rust_libs_root,
311 &args.web_libs_root,
312 &args.web_libs_published_root,
313 &args.rust_bins_root,
314 &args.rust_bin_artifacts_root,
315 ] {
316 let Some(parent) = path.parent().filter(|value| !value.as_os_str().is_empty()) else {
317 continue;
318 };
319 if parent.exists() {
320 continue;
321 }
322 let mut builder = std::fs::DirBuilder::new();
323 builder.recursive(true);
324 #[cfg(unix)]
325 {
326 use std::os::unix::fs::DirBuilderExt;
327 builder.mode(0o700);
328 }
329 builder
330 .create(parent)
331 .with_context(|| format!("creating runtime data directory {}", parent.display()))?;
332 }
333 Ok(())
334}
335
336pub async fn maintenance_guard(
337 bind: &str,
338 purpose: &str,
339) -> anyhow::Result<tokio::net::TcpListener> {
340 tokio::net::TcpListener::bind(bind).await.with_context(|| {
341 format!("binding maintenance lock {bind}; stop the running Kennedy server before {purpose}")
342 })
343}
344
345#[cfg(test)]
346mod tests {
347 use super::*;
348
349 #[test]
350 fn native_orchestration_remains_a_rust_backend_concern() {
351 assert_eq!(
352 std::any::type_name::<kcode_kennedy_orchestration::Session>(),
353 "kcode_kennedy_sessions::Session"
354 );
355 }
356
357 #[test]
358 fn dependency_closure_selects_current_internal_versions() {
359 let manifest = include_str!("../Cargo.toml");
360 assert!(manifest.contains("kcode-audio-ingress = \"0.7.6\""));
361 assert!(manifest.contains("kcode-kennedy-bootstrap-secrets = \"0.1.0\""));
362 assert!(manifest.contains("kcode-kennedy-orchestration = \"=0.3.2\""));
363 assert!(manifest.contains("kcode-kennedy-sessions = \"0.2.1\""));
364 assert!(manifest.contains("kcode-kennedy-telegram-runtime = \"0.3.2\""));
365 assert!(manifest.contains("kcode-session-history = \"0.1.13\""));
366 }
367
368 #[tokio::test]
369 async fn unified_dev_tools_service_opens_all_roots_and_routes_three_source_kinds() {
370 let directory = std::env::temp_dir().join(format!(
371 "kennedy-dev-tools-open-test-{}",
372 uuid::Uuid::new_v4()
373 ));
374 let rust_libraries = directory.join("kcode-rust-libs");
375 let web_libraries = directory.join("kcode-web-libs");
376 let web_publications = directory.join("kcode-web-libs-published");
377 let rust_binaries = directory.join("kcode-rust-bins");
378 let rust_binary_artifacts = directory.join("kcode-rust-bin-artifacts");
379 let service = kcode_dev_tools::Service::open(kcode_dev_tools::Config {
380 rust_libraries_root: rust_libraries.clone(),
381 web_libraries_root: web_libraries.clone(),
382 web_publications_root: web_publications.clone(),
383 rust_binaries_root: rust_binaries.clone(),
384 rust_binary_publications_root: rust_binary_artifacts.clone(),
385 crates_io_registry_token: "test-token".into(),
386 })
387 .unwrap();
388
389 assert_eq!(
390 service.web_libraries_root(),
391 std::fs::canonicalize(&web_libraries).unwrap()
392 );
393 assert_eq!(
394 service.web_publications_root(),
395 std::fs::canonicalize(&web_publications).unwrap()
396 );
397 for path in [
398 rust_libraries,
399 web_libraries,
400 web_publications,
401 rust_binaries,
402 rust_binary_artifacts,
403 ] {
404 assert!(
405 path.is_dir(),
406 "managed root was not created: {}",
407 path.display()
408 );
409 }
410 for (create, open, write, name, path, kind) in [
411 (
412 kcode_dev_tools::CREATE_RUST_LIB_TOOL,
413 kcode_dev_tools::OPEN_RUST_LIB_TOOL,
414 kcode_dev_tools::WRITE_FILE_FREEFORM_RUST_LIB_TOOL,
415 "kennedy-test-lib",
416 "src/extra.rs",
417 kcode_dev_tools::ManagedSourceKind::RustLibrary,
418 ),
419 (
420 kcode_dev_tools::CREATE_WEB_LIB_TOOL,
421 kcode_dev_tools::OPEN_WEB_LIB_TOOL,
422 kcode_dev_tools::WRITE_FILE_FREEFORM_WEB_LIB_TOOL,
423 "kennedy-test-web",
424 "extra.js",
425 kcode_dev_tools::ManagedSourceKind::WebLibrary,
426 ),
427 (
428 kcode_dev_tools::CREATE_RUST_BIN_TOOL,
429 kcode_dev_tools::OPEN_RUST_BIN_TOOL,
430 kcode_dev_tools::WRITE_FILE_FREEFORM_RUST_BIN_TOOL,
431 "kennedy-test-bin",
432 "src/extra.rs",
433 kcode_dev_tools::ManagedSourceKind::RustBinary,
434 ),
435 ] {
436 let created = service
437 .execute(
438 "create-session",
439 create,
440 serde_json::json!({"name":name}),
441 Vec::new(),
442 )
443 .await
444 .unwrap();
445 assert_eq!(created.snapshot.unwrap().kind, kind);
446 let written = service
447 .execute(
448 "create-session",
449 write,
450 serde_json::json!({
451 "name":name,
452 "path":path,
453 "contents":"// Kennedy managed source\n",
454 }),
455 Vec::new(),
456 )
457 .await
458 .unwrap();
459 assert_eq!(written.snapshot.unwrap().kind, kind);
460
461 let open_result = service
462 .execute(
463 "open-session",
464 open,
465 serde_json::json!({"name":name}),
466 Vec::new(),
467 )
468 .await
469 .unwrap();
470 assert_eq!(open_result.snapshot.unwrap().kind, kind);
471 }
472 let asset = service
473 .execute(
474 "create-session",
475 kcode_dev_tools::ATTACH_OBJECT_WEB_LIB_TOOL,
476 serde_json::json!({
477 "name":"kennedy-test-web",
478 "path":"assets/fonts/display.woff2",
479 "objectId":"pending:1",
480 }),
481 vec![vec![0, 159, 146, 150, 255]],
482 )
483 .await
484 .unwrap();
485 let snapshot = asset.snapshot.unwrap();
486 assert_eq!(
487 snapshot.kind,
488 kcode_dev_tools::ManagedSourceKind::WebLibrary
489 );
490 assert!(snapshot.text.contains("Asset: assets/fonts/display.woff2"));
491 assert!(snapshot.text.contains("Bytes: 5"));
492 assert!(!snapshot.text.contains("SHA-256:"));
493 assert_eq!(service.release("create-session").await.unwrap(), 3);
494 assert_eq!(service.release("open-session").await.unwrap(), 3);
495 drop(service);
496 std::fs::remove_dir_all(directory).unwrap();
497 }
498
499 #[tokio::test]
500 async fn occupied_kweb_address_prevents_server_from_opening_persistent_state() {
501 let directory =
502 std::env::temp_dir().join(format!("kennedy-server-lock-test-{}", uuid::Uuid::new_v4()));
503 std::fs::create_dir(&directory).unwrap();
504 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
505 let bind = listener.local_addr().unwrap().to_string();
506 let vault = directory.join("vault.age");
507 let kmap = directory.join("kweb");
508 let conversations = directory.join("conversations.sqlite3");
509 let telegram = directory.join("telegram.sqlite3");
510 let users = directory.join("users.sqlite3");
511 let tasks = directory.join("tasks.sqlite3");
512 let credits = directory.join("credits.sqlite3");
513 let audio_media = directory.join("audio-media");
514 let args = Args {
515 vault_path: vault.clone(),
516 command: None,
517 kweb_bind: bind,
518 kweb_root: kmap.clone(),
519 conversation_history_database: conversations.clone(),
520 session_directory: directory.join("sessions"),
521 session_history_file: directory.join("session-history.txt"),
522 telegram_database: telegram.clone(),
523 user_database: users.clone(),
524 task_board_database: tasks.clone(),
525 credits_database: credits.clone(),
526 audio_ingress_directory: audio_media.clone(),
527 intelligence_usage_directory: directory.join("intelligence-usage"),
528 rust_libs_root: directory.join("rust-libs"),
529 web_libs_root: directory.join("kcode-web-libs"),
530 web_libs_published_root: directory.join("kcode-web-libs-published"),
531 rust_bins_root: directory.join("kcode-rust-bins"),
532 rust_bin_artifacts_root: directory.join("kcode-rust-bin-artifacts"),
533 telegram_bootstrap_username: "@test".to_owned(),
534 telegram_max_voice_bytes: 1024,
535 audio_ingress_max_upload_bytes: 1024,
536 fixed: false,
537 };
538
539 let error = run_server(args, vault.clone()).await.unwrap_err();
540 assert!(error.to_string().contains("binding Kweb listener"));
541 assert!(!vault.exists());
542 assert!(!kmap.exists());
543 assert!(!conversations.exists());
544 assert!(!telegram.exists());
545 assert!(!users.exists());
546 assert!(!tasks.exists());
547 assert!(!credits.exists());
548 assert!(!audio_media.exists());
549 std::fs::remove_dir_all(directory).unwrap();
550 }
551}