qualia_cli/
daemon.rs

1use clap::{Args, Subcommand};
2use serde::{Deserialize, Serialize};
3use std::fs;
4use std::path::PathBuf;
5use std::process::{Command, Stdio};
6use std::time::Duration;
7use sysinfo::{Pid, Signal, System};
8
9pub const DEFAULT_DAEMON_PORT: u16 = 4242;
10const PID_FILE_NAME: &str = "daemon-service.json";
11const LOG_FILE_NAME: &str = "daemon-service.log";
12const ERR_LOG_FILE_NAME: &str = "daemon-service.err.log";
13
14#[derive(Args, Debug, Clone)]
15pub struct DaemonOpts {
16    /// Run in Development Mode (allows localhost origin and skips strict JWT pairing)
17    #[arg(long, global = true)]
18    pub dev: bool,
19    /// Local daemon port for the native bridge
20    #[arg(long, default_value_t = DEFAULT_DAEMON_PORT, global = true)]
21    pub port: u16,
22    /// Network Connectivity Profile (offline, metered, unmetered)
23    #[arg(long, default_value = "unmetered", global = true)]
24    pub net_mode: String,
25    /// Energy Circumstance Profile (strict, opportunistic, unlimited)
26    #[arg(long, default_value = "unlimited", global = true)]
27    pub energy_mode: String,
28    /// Fractal Sharding parallelism: number of 512MB cells to spin up
29    #[arg(long, default_value = "1", global = true)]
30    pub workers: u16,
31    /// Enable Sleep-Cycle Swarm AI Compute
32    #[arg(long, global = true)]
33    pub compute_swarm: bool,
34    /// Skip default graph seeding and startup ontologies (for docs/tests parity with WASM empty buffer)
35    #[arg(long, global = true)]
36    pub empty_graph: bool,
37}
38
39#[derive(Subcommand, Debug, Clone)]
40pub enum DaemonAction {
41    /// Run the graph daemon in the foreground (default when no subcommand is given)
42    Serve {
43        /// Hidden child mode used by `daemon start`
44        #[arg(long, hide = true)]
45        service_child: bool,
46    },
47    /// Start a detached daemon service on the loopback port (default 4242)
48    Start {
49        /// Replace an existing recorded service if its PID file is stale
50        #[arg(long)]
51        force: bool,
52    },
53    /// Stop the detached daemon service
54    Stop,
55    /// Report whether the detached daemon service is running
56    Status,
57    /// Inspect daemon health, port binding, and graph readiness
58    Doctor,
59}
60
61#[derive(Debug, Serialize, Deserialize)]
62struct DaemonServiceRecord {
63    pid: u32,
64    port: u16,
65    dev: bool,
66    started_at: String,
67    log_path: String,
68    workers: u16,
69    compute_swarm: bool,
70}
71
72pub async fn handle(action: &DaemonAction, opts: &DaemonOpts) {
73    match action {
74        DaemonAction::Serve { service_child } => {
75            if *service_child {
76                let _ = write_service_record(DaemonServiceRecord {
77                    pid: std::process::id(),
78                    port: opts.port,
79                    dev: opts.dev,
80                    started_at: chrono::Utc::now().to_rfc3339(),
81                    log_path: log_file_path().display().to_string(),
82                    workers: opts.workers,
83                    compute_swarm: opts.compute_swarm,
84                });
85            }
86            serve_foreground(opts, !*service_child).await;
87        }
88        DaemonAction::Start { force } => {
89            if let Err(err) = start_service(opts, *force) {
90                eprintln!("Failed to start daemon service: {err}");
91                std::process::exit(1);
92            }
93        }
94        DaemonAction::Stop => {
95            if let Err(err) = stop_service() {
96                eprintln!("Failed to stop daemon service: {err}");
97                std::process::exit(1);
98            }
99        }
100        DaemonAction::Status => {
101            if let Err(err) = print_status() {
102                eprintln!("Failed to inspect daemon service: {err}");
103                std::process::exit(1);
104            }
105        }
106        DaemonAction::Doctor => {
107            if let Err(err) = print_doctor(opts.port) {
108                eprintln!("Daemon doctor failed: {err}");
109                std::process::exit(1);
110            }
111        }
112    }
113}
114
115pub async fn serve_foreground(opts: &DaemonOpts, wait_for_ctrl_c: bool) {
116    let is_dev = opts.dev;
117    println!(
118        "Starting Qualia Native Loopback Server on 127.0.0.1:{}",
119        opts.port
120    );
121
122    println!("============================================================");
123    println!("Qualia-DB Zero-Allocation Native Local Daemon Booting...");
124    println!("============================================================");
125    println!("Network Mode: {}", opts.net_mode.to_uppercase());
126    println!("Energy Mode: {}", opts.energy_mode.to_uppercase());
127    println!("Fractal Shards: {} independent 512MB cells", opts.workers);
128    if opts.compute_swarm {
129        println!("Sleep-Cycle Swarm: ENABLED (Waiting for idle state...)");
130    }
131
132    if wait_for_ctrl_c {
133        tokio::spawn(async {
134            if let Ok(client) = reqwest::Client::builder()
135                .user_agent("qualia-cli-update-checker")
136                .build()
137            {
138                if let Ok(res) = client
139                    .get("https://crates.io/api/v1/crates/qualia-cli")
140                    .send()
141                    .await
142                {
143                    if let Ok(json) = res.json::<serde_json::Value>().await {
144                        if let Some(version) = json["crate"]["max_version"].as_str() {
145                            let current_version = env!("CARGO_PKG_VERSION");
146                            if version != current_version {
147                                println!("\n========================================");
148                                println!(
149                                    "A new version of qualia-cli (v{}) is available!",
150                                    version
151                                );
152                                println!("   You are currently running v{}", current_version);
153                                println!("   Run `cargo install qualia-cli --force` to update.");
154                                println!("========================================\n");
155                            }
156                        }
157                    }
158                }
159            }
160        });
161    }
162
163    if is_dev {
164        println!("WARNING: Running in DEV MODE. Trusting localhost origins.");
165    } else {
166        println!("Strict Origin Enforcement enabled: Trusting only mediaprophet.github.io");
167    }
168
169    let storage_dir = std::env::var("QUALIA_DATA_DIR").unwrap_or_else(|_| ".".to_string());
170    let vault = qualia_core_db::key_vault::KeyVault::load_or_generate(&storage_dir)
171        .expect("Failed to load KeyVault");
172    let vault_arc = std::sync::Arc::new(std::sync::Mutex::new(vault));
173    qualia_core_db::daemon::configure_daemon_topology(qualia_core_db::daemon::DaemonTopology {
174        worker_cells_configured: opts.workers,
175        compute_swarm_enabled: opts.compute_swarm,
176    });
177    qualia_core_db::daemon::start_local_daemon_with_options(
178        opts.port,
179        is_dev,
180        vault_arc,
181        opts.empty_graph,
182    )
183    .await;
184
185    if wait_for_ctrl_c {
186        println!("[Qualia Daemon] All subsystems active. Press Ctrl-C to shut down.");
187        tokio::signal::ctrl_c()
188            .await
189            .expect("Failed to install Ctrl-C handler");
190        println!("[Qualia Daemon] Shutdown signal received. Goodbye.");
191    } else {
192        println!(
193            "[Qualia Daemon] Background service active on 127.0.0.1:{}.",
194            opts.port
195        );
196        std::future::pending::<()>().await;
197    }
198}
199
200pub fn start_service(opts: &DaemonOpts, force: bool) -> Result<(), String> {
201    ensure_runtime_dir()?;
202
203    if let Some(record) = read_service_record()? {
204        if pid_is_running(record.pid) {
205            return Err(format!(
206                "daemon already running (pid {}, port {}). Use `qualia-cli daemon stop` first.",
207                record.pid, record.port
208            ));
209        }
210        if !force {
211            eprintln!(
212                "Removing stale daemon service record for pid {} on port {}.",
213                record.pid, record.port
214            );
215        }
216        let _ = clear_service_record();
217    }
218
219    let launched_pid = spawn_detached_service(opts)?;
220
221    for _ in 0..40 {
222        std::thread::sleep(Duration::from_millis(250));
223        if let Some(record) = read_service_record()? {
224            if record.port == opts.port && ping_daemon(opts.port).is_ok() {
225                println!(
226                    "Daemon service started on http://127.0.0.1:{} (pid {}).",
227                    record.port, record.pid
228                );
229                return Ok(());
230            }
231        }
232    }
233
234    let pid_hint = read_service_record()?
235        .map(|record| record.pid.to_string())
236        .or_else(|| launched_pid.map(|pid| pid.to_string()))
237        .unwrap_or_else(|| "unknown".to_string());
238    match ping_daemon(opts.port) {
239        Ok(_) => {
240            println!(
241                "Daemon service started on http://127.0.0.1:{} (pid {}).",
242                opts.port, pid_hint
243            );
244            Ok(())
245        }
246        Err(err) => Err(format!(
247            "spawned pid {pid_hint}, but health probe failed: {err}. Check {}",
248            err_log_file_path().display()
249        )),
250    }
251}
252
253pub fn stop_service() -> Result<(), String> {
254    let Some(record) = read_service_record()? else {
255        println!("Daemon service is not running.");
256        return Ok(());
257    };
258
259    if !pid_is_running(record.pid) {
260        clear_service_record()?;
261        println!(
262            "Removed stale daemon service record for pid {}.",
263            record.pid
264        );
265        return Ok(());
266    }
267
268    let mut system = System::new_all();
269    system.refresh_all();
270    let pid = Pid::from_u32(record.pid);
271    let Some(process) = system.process(pid) else {
272        clear_service_record()?;
273        println!(
274            "Removed stale daemon service record for pid {}.",
275            record.pid
276        );
277        return Ok(());
278    };
279
280    let terminated = process.kill_with(Signal::Term).unwrap_or(false) || process.kill();
281    if !terminated {
282        return Err(format!("unable to terminate pid {}", record.pid));
283    }
284
285    for _ in 0..15 {
286        if !pid_is_running(record.pid) {
287            clear_service_record()?;
288            println!("Stopped daemon service pid {}.", record.pid);
289            return Ok(());
290        }
291        std::thread::sleep(Duration::from_millis(200));
292    }
293
294    clear_service_record()?;
295    println!(
296        "Sent termination to pid {} and cleared the daemon service record.",
297        record.pid
298    );
299    Ok(())
300}
301
302pub fn print_status() -> Result<(), String> {
303    let Some(record) = read_service_record()? else {
304        println!("stopped");
305        return Ok(());
306    };
307
308    let running = pid_is_running(record.pid);
309    let health = ping_daemon(record.port).ok();
310    let health_label = if health.is_some() {
311        "healthy"
312    } else {
313        "unreachable"
314    };
315
316    if running {
317        println!(
318            "running pid={} port={} dev={} health={}",
319            record.pid, record.port, record.dev, health_label
320        );
321    } else {
322        println!(
323            "stale pid={} port={} dev={} health={}",
324            record.pid, record.port, record.dev, health_label
325        );
326    }
327    Ok(())
328}
329
330pub fn print_doctor(default_port: u16) -> Result<(), String> {
331    println!("Daemon doctor");
332    println!("  foreground       : qualia-cli daemon --dev");
333    println!(
334        "  background       : qualia-cli daemon start --dev --port {}",
335        default_port
336    );
337
338    match read_service_record()? {
339        Some(record) => {
340            println!(
341                "  service record   : pid={} port={} dev={}",
342                record.pid, record.port, record.dev
343            );
344            println!("  pid alive        : {}", pid_is_running(record.pid));
345            println!("  log file         : {}", record.log_path);
346
347            match ping_daemon(record.port) {
348                Ok(body) => {
349                    println!("  health           : ok");
350                    if let Some(version) = body.get("engine_version").and_then(|v| v.as_str()) {
351                        println!("  engine_version   : {version}");
352                    }
353                }
354                Err(err) => {
355                    println!("  health           : failed ({err})");
356                }
357            }
358        }
359        None => {
360            println!("  service record   : none");
361            match ping_daemon(default_port) {
362                Ok(_) => println!(
363                    "  health           : foreground daemon responding on port {default_port}"
364                ),
365                Err(_) => println!("  health           : service not running"),
366            }
367        }
368    }
369
370    Ok(())
371}
372
373pub fn ping_daemon(port: u16) -> Result<serde_json::Value, String> {
374    use std::io::{Read, Write};
375    use std::net::{SocketAddr, TcpStream};
376
377    let url = format!("http://127.0.0.1:{port}/health");
378    let addr: SocketAddr = format!("127.0.0.1:{port}")
379        .parse()
380        .map_err(|e| format!("parse {url}: {e}"))?;
381    let mut stream = TcpStream::connect_timeout(&addr, Duration::from_secs(3))
382        .map_err(|e| format!("connect {url}: {e}"))?;
383    stream
384        .set_read_timeout(Some(Duration::from_secs(3)))
385        .map_err(|e| e.to_string())?;
386    let request =
387        format!("GET /health HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nConnection: close\r\n\r\n");
388    stream
389        .write_all(request.as_bytes())
390        .map_err(|e| format!("write {url}: {e}"))?;
391
392    let mut body = String::new();
393    stream
394        .read_to_string(&mut body)
395        .map_err(|e| format!("read {url}: {e}"))?;
396    let json_start = body
397        .find('{')
398        .ok_or_else(|| format!("no JSON body in {url} response"))?;
399    serde_json::from_str(&body[json_start..]).map_err(|e| format!("decode {url}: {e}"))
400}
401
402fn spawn_detached_service(opts: &DaemonOpts) -> Result<Option<u32>, String> {
403    #[cfg(windows)]
404    {
405        spawn_detached_service_windows(opts)
406    }
407
408    #[cfg(not(windows))]
409    {
410        spawn_detached_service_portable(opts)
411    }
412}
413
414#[cfg(windows)]
415fn spawn_detached_service_windows(opts: &DaemonOpts) -> Result<Option<u32>, String> {
416    let current_exe = std::env::current_exe().map_err(|e| e.to_string())?;
417    let log = log_file_path();
418    let err = err_log_file_path();
419    let batch = runtime_dir().join("daemon-start.cmd");
420    let mut line = format!(
421        "@echo off\r\n\"{}\" daemon serve --port {} --net-mode {} --energy-mode {} --workers {} --service-child",
422        current_exe.display(),
423        opts.port,
424        opts.net_mode,
425        opts.energy_mode,
426        opts.workers
427    );
428    if opts.dev {
429        line.push_str(" --dev");
430    }
431    if opts.compute_swarm {
432        line.push_str(" --compute-swarm");
433    }
434    if opts.empty_graph {
435        line.push_str(" --empty-graph");
436    }
437    line.push_str(&format!(
438        " 1>> \"{}\" 2>> \"{}\"\r\n",
439        log.display(),
440        err.display()
441    ));
442    fs::write(&batch, line).map_err(|e| format!("write {}: {e}", batch.display()))?;
443
444    let mut command = Command::new("cmd");
445    command
446        .arg("/C")
447        .arg("start")
448        .arg("/B")
449        .arg("")
450        .arg(&batch)
451        .stdin(Stdio::null());
452
453    use std::os::windows::process::CommandExt;
454    const CREATE_NO_WINDOW: u32 = 0x08000000;
455    const CREATE_NEW_PROCESS_GROUP: u32 = 0x00000200;
456    command.creation_flags(CREATE_NO_WINDOW | CREATE_NEW_PROCESS_GROUP);
457
458    let child = command.spawn().map_err(|e| format!("spawn failed: {e}"))?;
459    Ok(Some(child.id()))
460}
461
462#[cfg(not(windows))]
463fn spawn_detached_service_portable(opts: &DaemonOpts) -> Result<Option<u32>, String> {
464    let current_exe = std::env::current_exe().map_err(|e| e.to_string())?;
465    let log = log_file_path();
466    let err = err_log_file_path();
467    let mut inner = format!(
468        "exec \"{}\" daemon serve --port {} --net-mode {} --energy-mode {} --workers {} --service-child",
469        current_exe.display(),
470        opts.port,
471        opts.net_mode,
472        opts.energy_mode,
473        opts.workers
474    );
475    if opts.dev {
476        inner.push_str(" --dev");
477    }
478    if opts.compute_swarm {
479        inner.push_str(" --compute-swarm");
480    }
481    if opts.empty_graph {
482        inner.push_str(" --empty-graph");
483    }
484    inner.push_str(&format!(
485        " >>\"{}\" 2>>\"{}\"",
486        log.display(),
487        err.display()
488    ));
489
490    let child = Command::new("sh")
491        .arg("-c")
492        .arg(&inner)
493        .stdin(Stdio::null())
494        .spawn()
495        .map_err(|e| format!("spawn failed: {e}"))?;
496    Ok(Some(child.id()))
497}
498
499fn ensure_runtime_dir() -> Result<(), String> {
500    fs::create_dir_all(runtime_dir()).map_err(|e| e.to_string())
501}
502
503fn runtime_dir() -> PathBuf {
504    state_dir().join("run")
505}
506
507fn state_dir() -> PathBuf {
508    if let Ok(dir) = std::env::var("QUALIA_DATA_DIR") {
509        return PathBuf::from(dir);
510    }
511    if let Ok(dir) = std::env::var("QUALIA_STORAGE_PATH") {
512        return PathBuf::from(dir);
513    }
514    if let Ok(dir) = std::env::current_dir() {
515        return dir.join(".qualia");
516    }
517    if let Ok(home) = std::env::var("HOME") {
518        return PathBuf::from(home).join(".qualia");
519    }
520    if let Ok(home) = std::env::var("USERPROFILE") {
521        return PathBuf::from(home).join(".qualia");
522    }
523    PathBuf::from(".qualia")
524}
525
526fn pid_file_path() -> PathBuf {
527    runtime_dir().join(PID_FILE_NAME)
528}
529
530fn log_file_path() -> PathBuf {
531    runtime_dir().join(LOG_FILE_NAME)
532}
533
534fn err_log_file_path() -> PathBuf {
535    runtime_dir().join(ERR_LOG_FILE_NAME)
536}
537
538fn read_service_record() -> Result<Option<DaemonServiceRecord>, String> {
539    let path = pid_file_path();
540    if !path.exists() {
541        return Ok(None);
542    }
543    let raw = fs::read_to_string(&path).map_err(|e| format!("read {}: {e}", path.display()))?;
544    let record =
545        serde_json::from_str(&raw).map_err(|e| format!("decode {}: {e}", path.display()))?;
546    Ok(Some(record))
547}
548
549fn write_service_record(record: DaemonServiceRecord) -> Result<(), String> {
550    ensure_runtime_dir()?;
551    let path = pid_file_path();
552    let raw = serde_json::to_string_pretty(&record).map_err(|e| e.to_string())?;
553    fs::write(&path, raw).map_err(|e| format!("write {}: {e}", path.display()))
554}
555
556fn clear_service_record() -> Result<(), String> {
557    let path = pid_file_path();
558    if path.exists() {
559        fs::remove_file(&path).map_err(|e| format!("remove {}: {e}", path.display()))?;
560    }
561    Ok(())
562}
563
564fn pid_is_running(pid: u32) -> bool {
565    let mut system = System::new_all();
566    system.refresh_all();
567    system.process(Pid::from_u32(pid)).is_some()
568}
qualia_cli/daemon.rs

qualia_cli/
daemon.rs
