//! Makima CLI - unified CLI for server, daemon, and task management. use std::io::{self, Read}; use std::path::Path; use std::sync::Arc; use makima::daemon::api::ApiClient; use makima::daemon::cli::{ Cli, CliConfig, Commands, ConfigCommand, DirectiveCommand, }; use makima::daemon::config::{DaemonConfig, RepoEntry}; use makima::daemon::db::LocalDb; use makima::daemon::error::DaemonError; use makima::daemon::setup; use makima::daemon::task::{TaskConfig, TaskManager}; use makima::daemon::ws::{DaemonCommand, WsClient}; use tokio::process::Command; use tokio::sync::mpsc; use tracing_subscriber::{fmt, prelude::*, EnvFilter}; #[tokio::main] async fn main() -> Result<(), Box> { let cli = Cli::parse_args(); match cli.command { Commands::Server(args) => run_server(args).await, Commands::Daemon(args) => run_daemon(args).await, Commands::Directive(cmd) => run_directive(cmd).await, Commands::Config(cmd) => run_config(cmd).await, } } /// Run the makima server. async fn run_server( args: makima::daemon::cli::ServerArgs, ) -> Result<(), Box> { // Initialize logging init_logging(&args.log_level, "text"); eprintln!("=== Makima Server Starting ==="); eprintln!("Port: {}", args.port); // Create app state. ML model paths are optional now — when none of // them are supplied (the slim Dockerfile case) Listen and Speak // websocket endpoints respond with "not configured" and everything // else works normally. let mut app_state = match ( args.parakeet_model_dir.as_deref(), args.parakeet_eou_dir.as_deref(), args.sortformer_model_path.as_deref(), args.chatterbox_model_dir.as_deref(), ) { (Some(p), Some(eou), Some(sf), Some(cb)) => { eprintln!("ML models configured (lazy load on first use)"); makima::server::state::AppState::new(p, eou, sf, cb) } (None, None, None, None) => { eprintln!("ML models NOT configured — Listen/Speak disabled"); makima::server::state::AppState::new_slim() } _ => { eprintln!( "WARNING: only some ML model paths provided. Pass all four \ (parakeet/parakeet_eou/sortformer/chatterbox) to enable ML, \ or none to run in slim mode. Continuing in slim mode." ); makima::server::state::AppState::new_slim() } }; // Connect to database if URL provided if let Some(ref db_url) = args.database_url { eprintln!("Connecting to database..."); let pool = makima::db::create_pool(db_url).await?; app_state = app_state.with_db_pool(pool); eprintln!("Database connected"); } let state = Arc::new(app_state); let addr = format!("0.0.0.0:{}", args.port); eprintln!("Starting server on {}", addr); makima::server::run_server(state, &addr).await?; Ok(()) } /// Run the makima daemon. async fn run_daemon( args: makima::daemon::cli::DaemonArgs, ) -> Result<(), Box> { eprintln!("=== Makima Daemon Starting ==="); // Check dependencies unless skipped if !args.skip_setup_check { eprintln!("[0/5] Checking dependencies..."); let dep_result = setup::check_dependencies().await; setup::print_dependency_summary(&dep_result); // Check for missing critical dependencies if !dep_result.claude.installed { let os = setup::OperatingSystem::detect(); setup::print_claude_install_instructions(os); std::process::exit(1); } if !dep_result.git.installed { let os = setup::OperatingSystem::detect(); setup::print_git_install_instructions(os); std::process::exit(1); } // Print git authentication warnings (non-fatal) setup::print_git_auth_warnings(&dep_result); } // Install Claude Code skills for makima commands eprintln!("[0.5/5] Installing Claude Code skills..."); if let Err(e) = makima::daemon::skill_installer::install_skills().await { eprintln!(" WARNING: Failed to install skills: {}", e); // Non-fatal: continue even if skill installation fails } // Build a temporary CLI struct for config loading let cli = makima::daemon::cli::daemon::DaemonArgs { config: args.config, repos_dir: args.repos_dir, worktrees_dir: args.worktrees_dir, server_url: args.server_url, api_key: args.api_key, max_tasks: args.max_tasks, log_level: args.log_level, bubblewrap: args.bubblewrap, skip_setup_check: args.skip_setup_check, }; // Load configuration with CLI overrides eprintln!("[1/5] Loading configuration..."); let config = match DaemonConfig::load_with_daemon_args(&cli) { Ok(cfg) => { eprintln!(" Config loaded: server={}", cfg.server.url); cfg } Err(e) => { eprintln!("Failed to load configuration: {}", e); eprintln!(); eprintln!("Use CLI flags:"); eprintln!(" makima daemon --server-url ws://localhost:8080 --api-key your-api-key"); eprintln!(); eprintln!("Or set environment variables:"); eprintln!(" MAKIMA_DAEMON_SERVER_URL=ws://localhost:8080"); eprintln!(" MAKIMA_API_KEY=your-api-key"); eprintln!(); eprintln!("Or create a config file: makima-daemon.toml"); std::process::exit(1); } }; // Initialize logging init_logging(&config.logging.level, &config.logging.format); eprintln!("[2/5] Logging initialized"); // Initialize local database eprintln!( "[3/5] Opening local database: {}", config.local_db.path.display() ); let local_db = Arc::new(std::sync::Mutex::new(LocalDb::open(&config.local_db.path)?)); eprintln!(" Database opened"); // Initialize worktree directories eprintln!("[4/5] Setting up directories..."); tokio::fs::create_dir_all(&config.worktree.base_dir).await?; tokio::fs::create_dir_all(&config.worktree.repos_dir).await?; tokio::fs::create_dir_all(&config.repos.home_dir).await?; eprintln!( " Worktree base: {}", config.worktree.base_dir.display() ); eprintln!(" Repos cache: {}", config.worktree.repos_dir.display()); eprintln!(" Home dir: {}", config.repos.home_dir.display()); // Auto-clone repositories if configured if !config.repos.auto_clone.is_empty() { eprintln!( " Auto-cloning {} repositories...", config.repos.auto_clone.len() ); for repo_entry in &config.repos.auto_clone { if let Err(e) = auto_clone_repo(repo_entry, &config.repos.home_dir).await { eprintln!(" WARNING: Failed to clone {}: {}", repo_entry.url(), e); } } } // Create channels for communication let (command_tx, mut command_rx) = mpsc::channel::(64); // Get machine info let machine_id = get_machine_id(); let hostname = get_hostname(); eprintln!(" Machine ID: {}", machine_id); eprintln!(" Hostname: {}", hostname); // Create WebSocket client eprintln!("[5/5] Connecting to server: {}", config.server.url); let mut ws_client = WsClient::new( config.server.clone(), machine_id, hostname, config.process.max_concurrent_tasks as i32, command_tx, ); // Get sender for task manager let ws_tx = ws_client.sender(); // Create task configuration let bubblewrap_config = if config.process.bubblewrap.enabled { Some(config.process.bubblewrap.clone()) } else { None }; // Derive HTTP API URL from WebSocket server URL (wss://... -> https://...) let api_url = config .server .url .replace("wss://", "https://") .replace("ws://", "http://"); let task_config = TaskConfig { max_concurrent_tasks: config.process.max_concurrent_tasks, max_tasks_per_contract: config.process.max_tasks_per_contract, worktree_base_dir: config.worktree.base_dir.clone(), env_vars: config.process.env_vars.clone(), claude_command: config.process.claude_command.clone(), claude_args: config.process.claude_args.clone(), claude_pre_args: config.process.claude_pre_args.clone(), enable_permissions: config.process.enable_permissions, disable_verbose: config.process.disable_verbose, bubblewrap: bubblewrap_config, api_url, api_key: config.server.api_key.clone(), heartbeat_commit_interval_secs: config.process.heartbeat_commit_interval_secs, checkpoint_patches: config.process.checkpoint_patches.clone(), }; // Create task manager with local database for crash recovery let task_manager = Arc::new(TaskManager::new(task_config, ws_tx.clone(), local_db)); // Recover any orphaned tasks from previous daemon run let recovered = task_manager.recover_orphaned_tasks().await; if !recovered.is_empty() { eprintln!(" Recovered {} orphaned tasks with intact worktrees", recovered.len()); } // Spawn command handler let task_manager_clone = task_manager.clone(); tokio::spawn(async move { tracing::info!("Command handler started, waiting for commands..."); while let Some(command) = command_rx.recv().await { tracing::info!("Received command from channel: {:?}", command); if let Err(e) = task_manager_clone.handle_command(command).await { tracing::error!("Failed to handle command: {}", e); } } tracing::info!("Command handler stopped"); }); // Spawn periodic worktree health check (every 60 seconds) let health_check_manager = task_manager.clone(); tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(60)); loop { interval.tick().await; let affected = health_check_manager.check_worktree_health().await; if !affected.is_empty() { tracing::info!( count = affected.len(), "Worktree health check detected missing worktrees - tasks marked for retry" ); } } }); // Handle shutdown signals let shutdown_signal = async { tokio::signal::ctrl_c() .await .expect("Failed to install Ctrl+C handler"); eprintln!("\nReceived shutdown signal"); }; eprintln!("=== Daemon running (Ctrl+C to stop) ==="); // Run WebSocket client with shutdown handling tokio::select! { result = ws_client.run() => { match result { Ok(()) => eprintln!("WebSocket client exited cleanly"), Err(DaemonError::AuthFailed(msg)) => { eprintln!("ERROR: Authentication failed: {}", msg); std::process::exit(1); } Err(e) => { eprintln!("ERROR: WebSocket client error: {}", e); std::process::exit(1); } } } _ = shutdown_signal => { eprintln!("Shutting down..."); } } // Gracefully shutdown all running Claude processes eprintln!("Terminating Claude processes..."); task_manager .shutdown_all_processes(std::time::Duration::from_secs(5)) .await; // Cleanup tracing::info!("Daemon stopped"); Ok(()) } /// Run directive commands. async fn run_directive( cmd: DirectiveCommand, ) -> Result<(), Box> { use makima::daemon::api::directive::*; match cmd { DirectiveCommand::List(args) => { let client = ApiClient::new(args.api_url, args.api_key)?; let result = client.list_directives().await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::Get(args) | DirectiveCommand::Status(args) => { let client = ApiClient::new(args.api_url, args.api_key)?; let result = client.get_directive(args.directive_id).await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::AddStep(args) => { let client = ApiClient::new(args.common.api_url, args.common.api_key)?; let depends_on: Vec = args .depends_on .map(|d| { d.split(',') .filter_map(|s| uuid::Uuid::parse_str(s.trim()).ok()) .collect() }) .unwrap_or_default(); let req = CreateStepRequest { name: args.name, description: args.description, task_plan: args.task_plan, depends_on, order_index: args.order_index, }; let result = client.directive_add_step(args.common.directive_id, req).await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::RemoveStep(args) => { let client = ApiClient::new(args.common.api_url, args.common.api_key)?; client.directive_remove_step(args.common.directive_id, args.step_id).await?; println!(r#"{{"success": true}}"#); } DirectiveCommand::SetDeps(args) => { let client = ApiClient::new(args.common.api_url, args.common.api_key)?; let depends_on: Vec = args .depends_on .split(',') .filter_map(|s| uuid::Uuid::parse_str(s.trim()).ok()) .collect(); let result = client .directive_set_deps(args.common.directive_id, args.step_id, depends_on) .await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::Start(args) => { let client = ApiClient::new(args.api_url, args.api_key)?; let result = client.directive_start(args.directive_id).await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::Pause(args) => { let client = ApiClient::new(args.api_url, args.api_key)?; let result = client.directive_pause(args.directive_id).await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::Advance(args) => { let client = ApiClient::new(args.api_url, args.api_key)?; let result = client.directive_advance(args.directive_id).await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::CompleteStep(args) => { let client = ApiClient::new(args.common.api_url, args.common.api_key)?; let result = client .directive_complete_step(args.common.directive_id, args.step_id) .await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::FailStep(args) => { let client = ApiClient::new(args.common.api_url, args.common.api_key)?; let result = client .directive_fail_step(args.common.directive_id, args.step_id) .await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::SkipStep(args) => { let client = ApiClient::new(args.common.api_url, args.common.api_key)?; let result = client .directive_skip_step(args.common.directive_id, args.step_id) .await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::BatchAddSteps(args) => { let client = ApiClient::new(args.common.api_url, args.common.api_key)?; let steps: serde_json::Value = serde_json::from_str(&args.json) .map_err(|e| format!("Invalid JSON: {}", e))?; let result = client .directive_batch_add_steps(args.common.directive_id, steps) .await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::Update(args) => { let client = ApiClient::new(args.common.api_url, args.common.api_key)?; let result = client .directive_update(args.common.directive_id, args.pr_url, args.pr_branch, args.status) .await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::Ask(args) => { let client = ApiClient::new(args.common.api_url.clone(), args.common.api_key.clone())?; eprintln!("Asking user: {}...", args.question); let choices = args .choices .map(|c| c.split(',').map(|s| s.trim().to_string()).collect()) .unwrap_or_default(); let result = client .supervisor_ask(&args.question, choices, args.context, args.timeout, args.phaseguard, args.multi_select, args.non_blocking, args.question_type) .await?; let mut response_value = result.0; // If the server returned still_pending, poll until we get a real response while response_value.get("stillPending").and_then(|v| v.as_bool()).unwrap_or(false) { // Extract question_id for polling let question_id_str = response_value.get("questionId") .and_then(|v| v.as_str()) .ok_or_else(|| { Box::::from( "Missing questionId in still_pending response" ) })?; let question_id: uuid::Uuid = question_id_str.parse().map_err(|e| { Box::::from( format!("Invalid questionId: {}", e) ) })?; eprintln!("Waiting for user response (polling)..."); let poll_result = client.supervisor_poll_question(question_id).await?; response_value = poll_result.0; } println!("{}", serde_json::to_string(&response_value)?); } DirectiveCommand::CreateOrder(args) => { // Validate order_type is spike or chore if args.order_type != "spike" && args.order_type != "chore" { eprintln!("Error: Only 'spike' and 'chore' order types are allowed. Got: '{}'", args.order_type); std::process::exit(1); } let client = ApiClient::new(args.common.api_url.clone(), args.common.api_key.clone())?; eprintln!("Creating order: {}...", args.title); let labels = args .labels .map(|l| { serde_json::Value::Array( l.split(',') .map(|s| serde_json::Value::String(s.trim().to_string())) .collect(), ) }) .unwrap_or_else(|| serde_json::json!([])); let req = makima::daemon::api::supervisor::CreateOrderRequest { title: args.title, description: args.description, priority: args.priority, order_type: args.order_type, labels, repository_url: None, }; let result = client.create_order(&req).await?; println!("{}", serde_json::to_string(&result.0)?); } DirectiveCommand::Verify(args) => { run_directive_verify(args).await?; } } Ok(()) } /// Run `makima directive verify` — checks that the current HEAD merges cleanly /// into `/`. Prints a JSON result and exits non-zero on conflict. /// /// Implementation uses `git merge-tree --write-tree` (Git ≥ 2.38), which performs /// the merge in-memory and lists conflicting paths without touching the working /// tree or creating any commits. async fn run_directive_verify( args: makima::daemon::cli::directive::VerifyArgs, ) -> Result<(), Box> { use std::process::Command; fn git(args: &[&str]) -> std::io::Result { Command::new("git").args(args).output() } let head_ref = args.head.as_deref().unwrap_or("HEAD").to_string(); let base_ref = format!("{}/{}", args.remote, args.base); if !args.skip_fetch { eprintln!("Fetching {} {}...", args.remote, args.base); let fetch = git(&["fetch", &args.remote, &args.base])?; if !fetch.status.success() { return Err(format!( "git fetch {} {} failed: {}", args.remote, args.base, String::from_utf8_lossy(&fetch.stderr) ) .into()); } } let head_rev = { let out = git(&["rev-parse", &head_ref])?; if !out.status.success() { return Err(format!( "git rev-parse {} failed: {}", head_ref, String::from_utf8_lossy(&out.stderr) ) .into()); } String::from_utf8_lossy(&out.stdout).trim().to_string() }; let base_rev = { let out = git(&["rev-parse", &base_ref])?; if !out.status.success() { return Err(format!( "git rev-parse {} failed (did you fetch?): {}", base_ref, String::from_utf8_lossy(&out.stderr) ) .into()); } String::from_utf8_lossy(&out.stdout).trim().to_string() }; eprintln!("Verifying merge: {} ({}) <- {} ({})", base_ref, &base_rev[..7.min(base_rev.len())], head_ref, &head_rev[..7.min(head_rev.len())]); let merge = Command::new("git") .args(["merge-tree", "--write-tree", "--name-only", "--no-messages", &base_rev, &head_rev]) .output()?; let stdout = String::from_utf8_lossy(&merge.stdout).to_string(); let stderr = String::from_utf8_lossy(&merge.stderr).to_string(); let success = merge.status.success(); let conflicting_files: Vec = if success { Vec::new() } else { stdout .lines() .skip(1) .filter(|l| !l.is_empty()) .map(|l| l.to_string()) .collect() }; let result = serde_json::json!({ "ok": success, "base": base_ref, "head": head_ref, "baseSha": base_rev, "headSha": head_rev, "conflictingFiles": conflicting_files, "goal": args.goal, }); println!("{}", serde_json::to_string(&result)?); if !success { eprintln!("\n[FAIL] Merge would conflict in {} file(s):", conflicting_files.len()); for f in &conflicting_files { eprintln!(" - {}", f); } if !stderr.is_empty() { eprintln!("\ngit stderr:\n{}", stderr); } eprintln!( "\nFix the conflicts before pushing. Typical workflow:\n \ git fetch {remote} {base}\n \ git merge {remote}/{base}\n \ # resolve conflicts, commit, then re-run `makima directive verify`", remote = args.remote, base = args.base, ); std::process::exit(1); } if let Some(goal) = &args.goal { eprintln!("\n[OK] No merge conflicts."); eprintln!("Reminder — directive goal:\n {}\n", goal); eprintln!("Confirm the diff (`git diff {}...HEAD`) actually delivers this goal before creating the PR.", base_ref); } else { eprintln!("[OK] No merge conflicts with {}.", base_ref); } Ok(()) } /// Run the TUI view command. /// Run config commands. async fn run_config(cmd: ConfigCommand) -> Result<(), Box> { match cmd { ConfigCommand::SetKey(args) => { let mut config = CliConfig::load(); config.api_key = Some(args.api_key); config.save()?; println!("API key saved to {:?}", CliConfig::config_path().unwrap_or_default()); Ok(()) } ConfigCommand::SetUrl(args) => { let mut config = CliConfig::load(); config.api_url = args.api_url; config.save()?; println!("API URL saved to {:?}", CliConfig::config_path().unwrap_or_default()); Ok(()) } ConfigCommand::Show => { let config = CliConfig::load(); println!("Configuration:"); println!(" API URL: {}", config.api_url); println!(" API Key: {}", config.api_key.as_ref().map(|k| { if k.len() > 10 { format!("{}...{}", &k[..6], &k[k.len()-4..]) } else { "***".to_string() } }).unwrap_or_else(|| "(not set)".to_string())); println!(); println!("Config file: {:?}", CliConfig::config_path().unwrap_or_default()); Ok(()) } ConfigCommand::Path => { if let Some(path) = CliConfig::config_path() { println!("{}", path.display()); } else { eprintln!("Could not determine config path"); } Ok(()) } } } /// Load contracts from API fn init_logging(level: &str, format: &str) { let filter = EnvFilter::try_from_default_env() .or_else(|_| EnvFilter::try_new(level)) .unwrap_or_else(|_| EnvFilter::new("info")); let subscriber = tracing_subscriber::registry().with(filter); if format == "json" { subscriber.with(fmt::layer().json()).init(); } else { subscriber.with(fmt::layer()).init(); } } fn get_machine_id() -> String { // Try to read machine-id from standard locations #[cfg(target_os = "linux")] { if let Ok(id) = std::fs::read_to_string("/etc/machine-id") { return id.trim().to_string(); } if let Ok(id) = std::fs::read_to_string("/var/lib/dbus/machine-id") { return id.trim().to_string(); } } #[cfg(target_os = "macos")] { // Use IOPlatformSerialNumber if let Ok(output) = std::process::Command::new("ioreg") .args(["-rd1", "-c", "IOPlatformExpertDevice"]) .output() { let stdout = String::from_utf8_lossy(&output.stdout); for line in stdout.lines() { if line.contains("IOPlatformUUID") { if let Some(uuid) = line.split('"').nth(3) { return uuid.to_string(); } } } } } // Fallback: generate a random ID and persist it let state_dir = dirs_next::data_local_dir() .unwrap_or_else(|| std::path::PathBuf::from(".")) .join("makima"); let machine_id_file = state_dir.join("machine-id"); if let Ok(id) = std::fs::read_to_string(&machine_id_file) { return id.trim().to_string(); } // Generate new ID let new_id = uuid::Uuid::new_v4().to_string(); std::fs::create_dir_all(&state_dir).ok(); std::fs::write(&machine_id_file, &new_id).ok(); new_id } fn get_hostname() -> String { hostname::get() .map(|h| h.to_string_lossy().to_string()) .unwrap_or_else(|_| "unknown".to_string()) } /// Auto-clone a repository to the home directory if it doesn't exist. async fn auto_clone_repo( repo_entry: &RepoEntry, home_dir: &Path, ) -> Result<(), Box> { let dir_name = repo_entry .dir_name() .ok_or("Could not determine directory name from URL")?; let target_dir = home_dir.join(&dir_name); // Check if already cloned if target_dir.exists() { eprintln!(" [skip] {} (already exists)", dir_name); return Ok(()); } let url = repo_entry.expanded_url(); eprintln!(" [clone] {} -> {}", url, target_dir.display()); // Build git clone command let mut args = vec!["clone".to_string()]; // Add shallow clone if requested if repo_entry.shallow() { args.push("--depth".to_string()); args.push("1".to_string()); } // Add branch if specified if let Some(branch) = repo_entry.branch() { args.push("--branch".to_string()); args.push(branch.to_string()); } args.push(url.clone()); args.push(target_dir.to_string_lossy().to_string()); // Run git clone let output = Command::new("git").args(&args).output().await?; if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); return Err(format!("git clone failed: {}", stderr).into()); } eprintln!(" [done] {}", dir_name); Ok(()) } /// dirs_next minimal replacement mod dirs_next { use std::path::PathBuf; pub fn data_local_dir() -> Option { #[cfg(target_os = "macos")] { std::env::var("HOME") .ok() .map(|h| PathBuf::from(h).join("Library").join("Application Support")) } #[cfg(target_os = "linux")] { std::env::var("XDG_DATA_HOME") .ok() .map(PathBuf::from) .or_else(|| { std::env::var("HOME") .ok() .map(|h| PathBuf::from(h).join(".local").join("share")) }) } #[cfg(target_os = "windows")] { std::env::var("LOCALAPPDATA").ok().map(PathBuf::from) } #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "windows")))] { None } } }