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_cache_safe_chain() {
359 let manifest = include_str!("../Cargo.toml");
360 assert!(manifest.contains("version = \"0.5.11\""));
361 assert!(manifest.contains("kcode-audio-ingress = \"0.7.6\""));
362 assert!(manifest.contains("kcode-dev-tools-chatend = \"=0.1.3\""));
363 assert!(manifest.contains("kcode-kennedy-bootstrap-secrets = \"0.1.0\""));
364 assert!(manifest.contains("kcode-kennedy-orchestration = \"=0.3.2\""));
365 assert!(manifest.contains("kcode-kennedy-sessions = \"=0.2.6\""));
366 assert!(manifest.contains("kcode-kennedy-telegram-runtime = \"0.3.2\""));
367 assert!(manifest.contains("kcode-kweb-context = \"=0.2.9\""));
368 assert!(manifest.contains("kcode-session-history = \"=0.1.15\""));
369 }
370
371 #[tokio::test]
372 async fn unified_dev_tools_service_opens_all_roots_and_routes_three_source_kinds() {
373 let directory = std::env::temp_dir().join(format!(
374 "kennedy-dev-tools-open-test-{}",
375 uuid::Uuid::new_v4()
376 ));
377 let rust_libraries = directory.join("kcode-rust-libs");
378 let web_libraries = directory.join("kcode-web-libs");
379 let web_publications = directory.join("kcode-web-libs-published");
380 let rust_binaries = directory.join("kcode-rust-bins");
381 let rust_binary_artifacts = directory.join("kcode-rust-bin-artifacts");
382 let service = kcode_dev_tools::Service::open(kcode_dev_tools::Config {
383 rust_libraries_root: rust_libraries.clone(),
384 web_libraries_root: web_libraries.clone(),
385 web_publications_root: web_publications.clone(),
386 rust_binaries_root: rust_binaries.clone(),
387 rust_binary_publications_root: rust_binary_artifacts.clone(),
388 crates_io_registry_token: "test-token".into(),
389 })
390 .unwrap();
391
392 assert_eq!(
393 service.web_libraries_root(),
394 std::fs::canonicalize(&web_libraries).unwrap()
395 );
396 assert_eq!(
397 service.web_publications_root(),
398 std::fs::canonicalize(&web_publications).unwrap()
399 );
400 for path in [
401 rust_libraries,
402 web_libraries,
403 web_publications,
404 rust_binaries,
405 rust_binary_artifacts,
406 ] {
407 assert!(
408 path.is_dir(),
409 "managed root was not created: {}",
410 path.display()
411 );
412 }
413 for (create, open, write, name, path, kind) in [
414 (
415 kcode_dev_tools::CREATE_RUST_LIB_TOOL,
416 kcode_dev_tools::OPEN_RUST_LIB_TOOL,
417 kcode_dev_tools::WRITE_FILE_FREEFORM_RUST_LIB_TOOL,
418 "kennedy-test-lib",
419 "src/extra.rs",
420 kcode_dev_tools::ManagedSourceKind::RustLibrary,
421 ),
422 (
423 kcode_dev_tools::CREATE_WEB_LIB_TOOL,
424 kcode_dev_tools::OPEN_WEB_LIB_TOOL,
425 kcode_dev_tools::WRITE_FILE_FREEFORM_WEB_LIB_TOOL,
426 "kennedy-test-web",
427 "extra.js",
428 kcode_dev_tools::ManagedSourceKind::WebLibrary,
429 ),
430 (
431 kcode_dev_tools::CREATE_RUST_BIN_TOOL,
432 kcode_dev_tools::OPEN_RUST_BIN_TOOL,
433 kcode_dev_tools::WRITE_FILE_FREEFORM_RUST_BIN_TOOL,
434 "kennedy-test-bin",
435 "src/extra.rs",
436 kcode_dev_tools::ManagedSourceKind::RustBinary,
437 ),
438 ] {
439 let created = service
440 .execute(
441 "create-session",
442 create,
443 serde_json::json!({"name":name}),
444 Vec::new(),
445 )
446 .await
447 .unwrap();
448 assert_eq!(created.snapshot.unwrap().kind, kind);
449 let written = service
450 .execute(
451 "create-session",
452 write,
453 serde_json::json!({
454 "name":name,
455 "path":path,
456 "contents":"// Kennedy managed source\n",
457 }),
458 Vec::new(),
459 )
460 .await
461 .unwrap();
462 assert_eq!(written.snapshot.unwrap().kind, kind);
463
464 let open_result = service
465 .execute(
466 "open-session",
467 open,
468 serde_json::json!({"name":name}),
469 Vec::new(),
470 )
471 .await
472 .unwrap();
473 assert_eq!(open_result.snapshot.unwrap().kind, kind);
474 }
475 let asset = service
476 .execute(
477 "create-session",
478 kcode_dev_tools::ATTACH_OBJECT_WEB_LIB_TOOL,
479 serde_json::json!({
480 "name":"kennedy-test-web",
481 "path":"assets/fonts/display.woff2",
482 "objectId":"pending:1",
483 }),
484 vec![vec![0, 159, 146, 150, 255]],
485 )
486 .await
487 .unwrap();
488 let snapshot = asset.snapshot.unwrap();
489 assert_eq!(
490 snapshot.kind,
491 kcode_dev_tools::ManagedSourceKind::WebLibrary
492 );
493 assert!(snapshot.text.contains("Asset: assets/fonts/display.woff2"));
494 assert!(snapshot.text.contains("Bytes: 5"));
495 assert!(!snapshot.text.contains("SHA-256:"));
496 assert_eq!(service.release("create-session").await.unwrap(), 3);
497 assert_eq!(service.release("open-session").await.unwrap(), 3);
498 drop(service);
499 std::fs::remove_dir_all(directory).unwrap();
500 }
501
502 #[tokio::test]
503 async fn occupied_kweb_address_prevents_server_from_opening_persistent_state() {
504 let directory =
505 std::env::temp_dir().join(format!("kennedy-server-lock-test-{}", uuid::Uuid::new_v4()));
506 std::fs::create_dir(&directory).unwrap();
507 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
508 let bind = listener.local_addr().unwrap().to_string();
509 let vault = directory.join("vault.age");
510 let kmap = directory.join("kweb");
511 let conversations = directory.join("conversations.sqlite3");
512 let telegram = directory.join("telegram.sqlite3");
513 let users = directory.join("users.sqlite3");
514 let tasks = directory.join("tasks.sqlite3");
515 let credits = directory.join("credits.sqlite3");
516 let audio_media = directory.join("audio-media");
517 let args = Args {
518 vault_path: vault.clone(),
519 command: None,
520 kweb_bind: bind,
521 kweb_root: kmap.clone(),
522 conversation_history_database: conversations.clone(),
523 session_directory: directory.join("sessions"),
524 session_history_file: directory.join("session-history.txt"),
525 telegram_database: telegram.clone(),
526 user_database: users.clone(),
527 task_board_database: tasks.clone(),
528 credits_database: credits.clone(),
529 audio_ingress_directory: audio_media.clone(),
530 intelligence_usage_directory: directory.join("intelligence-usage"),
531 rust_libs_root: directory.join("rust-libs"),
532 web_libs_root: directory.join("kcode-web-libs"),
533 web_libs_published_root: directory.join("kcode-web-libs-published"),
534 rust_bins_root: directory.join("kcode-rust-bins"),
535 rust_bin_artifacts_root: directory.join("kcode-rust-bin-artifacts"),
536 telegram_bootstrap_username: "@test".to_owned(),
537 telegram_max_voice_bytes: 1024,
538 audio_ingress_max_upload_bytes: 1024,
539 fixed: false,
540 };
541
542 let error = run_server(args, vault.clone()).await.unwrap_err();
543 assert!(error.to_string().contains("binding Kweb listener"));
544 assert!(!vault.exists());
545 assert!(!kmap.exists());
546 assert!(!conversations.exists());
547 assert!(!telegram.exists());
548 assert!(!users.exists());
549 assert!(!tasks.exists());
550 assert!(!credits.exists());
551 assert!(!audio_media.exists());
552 std::fs::remove_dir_all(directory).unwrap();
553 }
554}