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