#[path = "daemon_run.rs"]
mod daemon_run;
pub(crate) use daemon_run::run_daemon;
use serde::Deserialize;
use crate::formatters::session::short_id;
use crate::types::SessionRow;
pub(crate) fn git_root_for(workdir: &str) -> String {
let out = std::process::Command::new("git")
.arg("-C")
.arg(workdir)
.args(["rev-parse", "--show-toplevel"])
.output()
.ok()
.filter(|o| o.status.success());
match out {
Some(o) => String::from_utf8_lossy(&o.stdout).trim().to_string(),
None => workdir.to_string(),
}
}
pub(crate) fn group_by_git_root<'a>(
sessions: &'a [SessionRow],
cwd_root: &str,
) -> Vec<(String, Vec<&'a SessionRow>)> {
use std::collections::{BTreeMap, HashMap};
let mut cache: HashMap<&str, String> = HashMap::new();
let mut map: BTreeMap<String, Vec<&SessionRow>> = BTreeMap::new();
for s in sessions {
let key = cache
.entry(s.workdir.as_str())
.or_insert_with(|| git_root_for(&s.workdir))
.clone();
map.entry(key).or_default().push(s);
}
let mut groups: Vec<(String, Vec<&SessionRow>)> = map.into_iter().collect();
if let Some(pos) = groups.iter().position(|(k, _)| k == cwd_root) {
let cwd_group = groups.remove(pos);
groups.insert(0, cwd_group);
}
groups
}
pub(crate) async fn print_status(client: &reqwest::Client, url: &str) -> anyhow::Result<()> {
println!("daemon: ok");
#[derive(Deserialize)]
struct Body {
sessions: Vec<SessionRow>,
}
let body: Body = client
.get(format!("{url}/sessions"))
.send()
.await?
.error_for_status()?
.json()
.await?;
let cwd = std::env::current_dir()
.map(|p| p.to_string_lossy().into_owned())
.unwrap_or_default();
let cwd_root = git_root_for(&cwd);
let groups = group_by_git_root(&body.sessions, &cwd_root);
for (root, sessions) in &groups {
let marker = if root == &cwd_root { " *" } else { "" };
println!("---- {} ({} session(s)){marker} ----", root, sessions.len());
for s in sessions {
let status = s.status.as_str().unwrap_or("unknown");
println!(
" {} {} {} ({} delegations)",
short_id(&s.id),
status,
s.workdir,
s.active_delegations
);
}
}
if trusty_mpm::telegram::resolve_token("TELEGRAM_BOT_TOKEN").is_some() {
println!("Telegram bot active");
}
Ok(())
}
pub(crate) async fn daemon_healthy(client: &reqwest::Client, url: &str) -> bool {
let health_ok = match client.get(format!("{url}/health")).send().await {
Ok(r) => r.status().is_success(),
Err(_) => return false,
};
if !health_ok {
return false;
}
match client.get(format!("{url}/sessions")).send().await {
Ok(r) => r.status().is_success(),
Err(_) => false,
}
}
pub(crate) async fn start(client: &reqwest::Client, url: &str) -> anyhow::Result<()> {
let lock_url = trusty_mpm::core::resolve_daemon_url(None);
let check_url = if lock_url != trusty_mpm::core::DEFAULT_DAEMON_URL
|| url == trusty_mpm::core::DEFAULT_DAEMON_URL
{
lock_url.clone()
} else {
url.to_string()
};
if daemon_healthy(client, &check_url).await {
println!("Daemon already running on {check_url}");
return print_status(client, &check_url).await;
}
if let Some(label) = crate::commands::launchd_probe::daemon_launchd_label() {
anyhow::bail!(crate::commands::launchd_probe::cli_spawn_refusal_hint(
&label
));
}
let root = trusty_mpm::core::paths::FrameworkPaths::default().root;
std::fs::create_dir_all(&root)?;
let log_path = root.join("daemon.log");
let stdout = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&log_path)?;
let stderr = stdout.try_clone()?;
let lock_path = trusty_mpm::core::lock_file_path();
let _ = std::fs::remove_file(&lock_path);
let exe = std::env::current_exe()?;
let stable_dir = dirs::home_dir().unwrap_or_else(|| std::path::PathBuf::from("/"));
std::process::Command::new(&exe)
.arg("daemon")
.current_dir(&stable_dir)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::from(stdout))
.stderr(std::process::Stdio::from(stderr))
.spawn()?;
print!("Starting daemon... ");
use std::io::Write as _;
std::io::stdout().flush().ok();
let mut healthy = false;
let mut actual_url = url.to_string();
for _ in 0..10 {
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
actual_url = trusty_mpm::core::resolve_daemon_url(None);
if daemon_healthy(client, &actual_url).await {
healthy = true;
break;
}
}
if healthy {
println!("done");
if actual_url != url {
println!("(listening on {actual_url})");
}
} else {
println!("failed");
println!(
"daemon did not become healthy within 5s; see {}",
log_path.display()
);
return Ok(());
}
print_status(client, &actual_url).await
}
pub(crate) async fn restart(client: &reqwest::Client, url: &str) -> anyhow::Result<()> {
if let Some(label) = crate::commands::launchd_probe::daemon_launchd_label() {
anyhow::bail!(crate::commands::launchd_probe::cli_spawn_refusal_hint(
&label
));
}
if daemon_healthy(client, url).await {
print!("Stopping daemon... ");
use std::io::Write as _;
std::io::stdout().flush().ok();
std::process::Command::new("pkill")
.args(["-f", "tm daemon"])
.status()
.ok();
std::process::Command::new("pkill")
.args(["-f", "trusty-mpm daemon"])
.status()
.ok();
for _ in 0..6 {
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
if !daemon_healthy(client, url).await {
break;
}
}
println!("done");
}
start(client, url).await
}
pub(crate) async fn stop_daemon() -> anyhow::Result<()> {
use std::time::{Duration, Instant};
let targets = find_daemon_pids();
if targets.is_empty() {
anyhow::bail!("No daemon running");
}
println!(
"Stopping trusty-mpm daemon ({} process(es): {:?})…",
targets.len(),
targets
);
if let Some(label) = crate::commands::launchd_probe::daemon_launchd_label() {
println!(
"note: launchd unit `{label}` owns this daemon and will NOT respawn it \
(KeepAlive.SuccessfulExit=false). Bring it back with `{}` — issue #4230.",
crate::commands::launchd_probe::daemon_restart_command_for(Some(&label))
);
}
for pid in &targets {
let _ = send_signal(*pid, "TERM");
}
let deadline = Instant::now() + Duration::from_secs(5);
loop {
tokio::time::sleep(Duration::from_millis(100)).await;
let any_alive = targets.iter().any(|p| pid_alive(*p));
if !any_alive {
println!("Daemon stopped");
cleanup_lock_file();
return Ok(());
}
if Instant::now() >= deadline {
break;
}
}
let stragglers: Vec<u32> = targets.iter().copied().filter(|p| pid_alive(*p)).collect();
if !stragglers.is_empty() {
println!(
"{} process(es) ignored SIGTERM — sending SIGKILL: {:?}",
stragglers.len(),
stragglers
);
for pid in &stragglers {
let _ = send_signal(*pid, "KILL");
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
if targets.iter().any(|p| pid_alive(*p)) {
println!("Daemon may still be shutting down");
} else {
println!("Daemon stopped");
cleanup_lock_file();
}
Ok(())
}
pub(crate) fn cleanup_lock_file() {
let path = trusty_mpm::core::lock_file_path();
let _ = std::fs::remove_file(&path);
let root = trusty_mpm::core::paths::FrameworkPaths::default().root;
super::guided_autostart::remove_autostart_pidfile(&root);
}
pub(crate) fn find_daemon_pids() -> Vec<u32> {
use sysinfo::{ProcessRefreshKind, RefreshKind, System};
let mut sys = System::new_with_specifics(
RefreshKind::nothing().with_processes(ProcessRefreshKind::nothing()),
);
sys.refresh_processes(sysinfo::ProcessesToUpdate::All, true);
let me = std::process::id();
let mut out = Vec::new();
for (pid, proc_) in sys.processes() {
let raw = pid.as_u32();
if raw == me {
continue;
}
let name = proc_.name().to_string_lossy();
let is_tm_binary = trusty_mpm::core::own_binary_names::OWN_BINARY_NAMES
.iter()
.any(|n| name == *n);
if !is_tm_binary {
continue;
}
let is_daemon = proc_.cmd().iter().any(|a| a.to_string_lossy() == "daemon");
if is_daemon {
out.push(raw);
}
}
out
}
#[cfg(unix)]
pub(crate) fn send_signal(pid: u32, sig: &str) -> std::io::Result<()> {
let status = std::process::Command::new("kill")
.arg(format!("-{sig}"))
.arg(pid.to_string())
.status()?;
if !status.success() {
return Err(std::io::Error::other(format!(
"kill -{sig} {pid} exited {status}"
)));
}
Ok(())
}
#[cfg(not(unix))]
pub(crate) fn send_signal(_pid: u32, _sig: &str) -> std::io::Result<()> {
Err(std::io::Error::other(
"signals unsupported on this platform",
))
}
#[cfg(unix)]
pub(crate) fn pid_alive(pid: u32) -> bool {
std::process::Command::new("kill")
.arg("-0")
.arg(pid.to_string())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
#[cfg(not(unix))]
pub(crate) fn pid_alive(_pid: u32) -> bool {
true
}
#[cfg(test)]
mod tests {
use super::*;
fn make_session(workdir: &str) -> SessionRow {
SessionRow {
id: serde_json::json!({"0": "00000000-0000-0000-0000-000000000000"}),
workdir: workdir.to_string(),
status: serde_json::json!("active"),
active_delegations: 0,
}
}
#[test]
fn group_sessions_by_git_root_no_git() {
let sessions = vec![make_session("/tmp/no-git-dir")];
let groups = group_by_git_root(&sessions, "/tmp/no-git-dir");
assert_eq!(groups.len(), 1, "one group for the single session");
assert_eq!(groups[0].0, "/tmp/no-git-dir");
assert_eq!(groups[0].1.len(), 1);
}
#[test]
fn group_sessions_by_git_root_cwd_first() {
let sessions = vec![
make_session("/tmp/alpha-dir"),
make_session("/tmp/beta-dir"),
];
let groups = group_by_git_root(&sessions, "/tmp/beta-dir");
assert_eq!(groups[0].0, "/tmp/beta-dir", "cwd_root group must be first");
}
#[test]
fn group_sessions_same_root_together() {
let sessions = vec![make_session("/tmp/same-dir"), make_session("/tmp/same-dir")];
let groups = group_by_git_root(&sessions, "/tmp/same-dir");
assert_eq!(groups.len(), 1, "same root → one group");
assert_eq!(groups[0].1.len(), 2, "two sessions in the group");
}
#[test]
fn git_root_for_nonexistent_path_returns_workdir() {
let path = "/tmp/definitely-not-a-git-repo-xyzzy-12345";
let result = git_root_for(path);
assert_eq!(result, path, "non-git path must fall back to workdir");
}
}