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.12\""));
354 assert!(manifest.contains("kcode-audio-ingress = \"0.7.6\""));
355 assert!(manifest.contains("kcode-dev-tools-chatend = \"0.1.3\""));
356 assert!(manifest.contains("kcode-kennedy-bootstrap-secrets = \"0.1.0\""));
357 assert!(manifest.contains("kcode-kennedy-maintenance-guard = \"0.1.0\""));
358 assert!(manifest.contains("kcode-kennedy-orchestration = \"0.3.3\""));
359 assert!(!manifest.contains("kcode-kennedy-orchestration = \"=0.3.3\""));
360 assert!(manifest.contains("kcode-kennedy-sessions = \"0.2.7\""));
361 assert!(!manifest.contains("kcode-kennedy-sessions = \"=0.2.7\""));
362 assert!(manifest.contains("kcode-kennedy-telegram-runtime = \"0.3.3\""));
363 assert!(manifest.contains("kcode-kweb-context = \"0.2.9\""));
364 assert!(manifest.contains("kcode-session-history = \"0.1.15\""));
365 assert!(manifest.contains("kcode-telegram-identity = \"0.1.8\""));
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}