use std::fs::File;
use std::path::Path;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
use tracing::{error, info, warn};
use super::{config, daemon, events, hierarchy, inbox, layout, team_config_path};
use crate::team::daemon::health::binary_freshness::{
BinaryFreshness, STALE_MANUAL_RECOVERY_COMMAND, STALE_RECOVERY_COMMAND,
evaluate_binary_freshness,
};
use crate::team::merge::{RootDirtyState, inspect_root_dirty_state};
use crate::tmux;
pub(crate) const LOG_ROTATION_BYTES: u64 = 5 * 1024 * 1024;
const LOG_ROTATION_KEEP: usize = 3;
pub(super) const DAEMON_SHUTDOWN_GRACE_PERIOD: Duration = Duration::from_secs(5);
const DAEMON_SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(100);
const WATCHDOG_POLL_INTERVAL: Duration = Duration::from_millis(200);
const WATCHDOG_INITIAL_BACKOFF_SECS: u64 = 1;
const WATCHDOG_MAX_BACKOFF_SECS: u64 = 30;
const WATCHDOG_CIRCUIT_BREAKER_THRESHOLD: usize = 5;
const WATCHDOG_CIRCUIT_BREAKER_WINDOW_SECS: u64 = 60;
const DAEMON_CHILD_PID_FILE: &str = "daemon-child.pid";
#[allow(dead_code)] pub(crate) const DAEMON_EXIT_CATEGORY_TRANSIENT: &str = "transient";
pub(crate) const DAEMON_EXIT_CATEGORY_UNRECOVERABLE: &str = "unrecoverable";
pub(crate) const DAEMON_EXIT_CATEGORY_UNKNOWN: &str = "unknown";
#[cfg(unix)]
static WATCHDOG_SHUTDOWN_REQUESTED: AtomicBool = AtomicBool::new(false);
pub(crate) fn watchdog_pid_path(project_root: &Path) -> PathBuf {
project_root.join(".batty").join("daemon.pid")
}
pub(crate) fn daemon_log_path(project_root: &Path) -> PathBuf {
project_root.join(".batty").join("daemon.log")
}
fn rotated_log_path(path: &Path, generation: usize) -> PathBuf {
PathBuf::from(format!("{}.{}", path.display(), generation))
}
pub(crate) fn rotate_log_if_needed(path: &Path) -> Result<()> {
let len = match std::fs::metadata(path) {
Ok(metadata) => metadata.len(),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(error) => {
return Err(error).with_context(|| format!("failed to stat {}", path.display()));
}
};
if len <= LOG_ROTATION_BYTES {
return Ok(());
}
let oldest = rotated_log_path(path, LOG_ROTATION_KEEP);
if oldest.exists() {
std::fs::remove_file(&oldest)
.with_context(|| format!("failed to remove {}", oldest.display()))?;
}
for generation in (1..LOG_ROTATION_KEEP).rev() {
let source = rotated_log_path(path, generation);
if !source.exists() {
continue;
}
let destination = rotated_log_path(path, generation + 1);
std::fs::rename(&source, &destination).with_context(|| {
format!(
"failed to rotate {} to {}",
source.display(),
destination.display()
)
})?;
}
let rotated = rotated_log_path(path, 1);
std::fs::rename(path, &rotated).with_context(|| {
format!(
"failed to rotate {} to {}",
path.display(),
rotated.display()
)
})?;
Ok(())
}
pub(crate) fn open_log_for_append(path: &Path) -> Result<File> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
rotate_log_if_needed(path)?;
File::options()
.append(true)
.create(true)
.open(path)
.with_context(|| format!("failed to open log file: {}", path.display()))
}
fn daemon_spawn_args(root_str: &str, resume: bool) -> Vec<String> {
let mut args = vec![
"-v".to_string(),
"daemon".to_string(),
"--project-root".to_string(),
root_str.to_string(),
];
if resume {
args.push("--resume".to_string());
}
args
}
fn watchdog_spawn_args(root_str: &str, resume: bool) -> Vec<String> {
let mut args = vec![
"-v".to_string(),
"watchdog".to_string(),
"--project-root".to_string(),
root_str.to_string(),
];
if resume {
args.push("--resume".to_string());
}
args
}
pub(crate) fn daemon_state_path(project_root: &Path) -> PathBuf {
project_root.join(".batty").join("daemon-state.json")
}
pub(crate) fn watchdog_state_path(project_root: &Path) -> PathBuf {
project_root.join(".batty").join("watchdog-state.json")
}
pub(crate) fn daemon_child_pid_path(project_root: &Path) -> PathBuf {
project_root.join(".batty").join(DAEMON_CHILD_PID_FILE)
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) struct PersistedWatchdogState {
#[serde(default)]
pub restart_count: u32,
#[serde(default)]
pub crash_timestamps: Vec<u64>,
#[serde(default)]
pub circuit_breaker_tripped: bool,
#[serde(default)]
pub child_pid: Option<u32>,
#[serde(default)]
pub current_backoff_secs: Option<u64>,
#[serde(default)]
pub last_exit_reason: Option<String>,
#[serde(default)]
pub last_exit_category: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct DaemonExitObservation {
reason: String,
exit_category: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StaleDaemonRestartStatus {
FreshNoop,
DryRunReady,
Restarted,
RefusedDirty,
RefusedActiveMerge,
RefusedMissingReleaseBinary,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StaleDaemonRestartReport {
pub status: StaleDaemonRestartStatus,
pub message: String,
pub commands: Vec<String>,
pub dirty_paths: Vec<String>,
pub freshness: Option<BinaryFreshness>,
}
impl StaleDaemonRestartReport {
pub fn render(&self) -> String {
let mut lines = vec![self.message.clone()];
if !self.dirty_paths.is_empty() {
lines.push(format!(
"Dirty source paths: {}",
self.dirty_paths.join(", ")
));
}
if !self.commands.is_empty() {
lines.push("Planned commands:".to_string());
lines.extend(self.commands.iter().map(|command| format!("- {command}")));
}
lines.join("\n")
}
}
pub(crate) fn classify_daemon_exit_reason(reason: &str) -> &'static str {
let reason = reason.to_ascii_lowercase();
if reason.contains("daemon startup pre-flight failed")
|| reason.contains("no team config found")
|| reason.contains("tmux server died")
|| reason.contains("tmux session disappeared")
|| (reason.contains("tmux session '") && reason.contains("not found"))
{
DAEMON_EXIT_CATEGORY_UNRECOVERABLE
} else {
DAEMON_EXIT_CATEGORY_UNKNOWN
}
}
pub fn restart_daemon_if_stale(
project_root: &Path,
dry_run: bool,
) -> Result<StaleDaemonRestartReport> {
let binary_path = std::env::current_exe().context("failed to resolve current executable")?;
restart_daemon_if_stale_with_binary_path(project_root, &binary_path, dry_run)
}
pub(crate) fn restart_daemon_if_stale_with_binary_path(
project_root: &Path,
binary_path: &Path,
dry_run: bool,
) -> Result<StaleDaemonRestartReport> {
let decision = safe_daemon_restart_decision_with_binary_path(project_root, binary_path)?;
if dry_run || decision.status != StaleDaemonRestartStatus::DryRunReady {
return Ok(decision);
}
let Some(freshness) = decision.freshness.clone() else {
return Ok(decision);
};
let commands = decision.commands.clone();
rebuild_reinstall_and_resign_batty(project_root)?;
if !request_graceful_daemon_shutdown(project_root, DAEMON_SHUTDOWN_GRACE_PERIOD) {
bail!(
"daemon did not stop gracefully; refusing to force restart after reinstall. Manual fallback: {}",
STALE_MANUAL_RECOVERY_COMMAND
);
}
let pid = spawn_watchdog(project_root, true)?;
Ok(StaleDaemonRestartReport {
status: StaleDaemonRestartStatus::Restarted,
message: format!(
"Daemon binary was stale by {} source commit(s); rebuilt, reinstalled, and restarted watchdog pid {pid}.",
freshness.commits_behind
),
commands,
dirty_paths: Vec::new(),
freshness: Some(freshness),
})
}
pub(crate) fn safe_daemon_restart_decision(
project_root: &Path,
) -> Result<StaleDaemonRestartReport> {
let binary_path = std::env::current_exe().context("failed to resolve current executable")?;
safe_daemon_restart_decision_with_binary_path(project_root, &binary_path)
}
pub(crate) fn safe_daemon_restart_decision_with_binary_path(
project_root: &Path,
binary_path: &Path,
) -> Result<StaleDaemonRestartReport> {
let freshness = evaluate_binary_freshness(binary_path, project_root)?;
let Some(freshness) = freshness else {
return Ok(StaleDaemonRestartReport {
status: StaleDaemonRestartStatus::FreshNoop,
message: format!(
"Daemon binary freshness is unavailable for {}; no restart attempted.",
binary_path.display()
),
commands: Vec::new(),
dirty_paths: Vec::new(),
freshness: None,
});
};
if freshness.fresh {
return Ok(StaleDaemonRestartReport {
status: StaleDaemonRestartStatus::FreshNoop,
message: "Daemon binary is fresh; no restart needed.".to_string(),
commands: Vec::new(),
dirty_paths: Vec::new(),
freshness: Some(freshness),
});
}
if merge_lock_path(project_root).exists() {
return Ok(StaleDaemonRestartReport {
status: StaleDaemonRestartStatus::RefusedActiveMerge,
message: "Refusing stale-daemon restart because a merge is active.".to_string(),
commands: Vec::new(),
dirty_paths: Vec::new(),
freshness: Some(freshness),
});
}
let root_dirty = inspect_root_dirty_state(project_root)?;
if !root_dirty.source_paths.is_empty() {
return Ok(refused_dirty_restart_report(freshness, &root_dirty));
}
let release_binary = release_batty_path(project_root);
if !release_binary.is_file() {
return Ok(StaleDaemonRestartReport {
status: StaleDaemonRestartStatus::RefusedMissingReleaseBinary,
message: format!(
"Refusing stale-daemon restart because release binary is missing at {}. Run `cargo build --release` before retrying.",
release_binary.display()
),
commands: Vec::new(),
dirty_paths: Vec::new(),
freshness: Some(freshness),
});
}
let commands = stale_daemon_restart_commands();
Ok(StaleDaemonRestartReport {
status: StaleDaemonRestartStatus::DryRunReady,
message: format!(
"Daemon binary is stale by {} source commit(s); safe restart path is ready. Runtime-only dirty paths are tolerated.",
freshness.commits_behind
),
commands,
dirty_paths: Vec::new(),
freshness: Some(freshness),
})
}
fn refused_dirty_restart_report(
freshness: BinaryFreshness,
root_dirty: &RootDirtyState,
) -> StaleDaemonRestartReport {
StaleDaemonRestartReport {
status: StaleDaemonRestartStatus::RefusedDirty,
message: format!(
"Refusing stale-daemon restart because source edits are present. Clean, commit, or stash these paths, then run `{}`. Manual fallback after cleaning: `{}`",
STALE_RECOVERY_COMMAND, STALE_MANUAL_RECOVERY_COMMAND
),
commands: Vec::new(),
dirty_paths: root_dirty.source_paths.clone(),
freshness: Some(freshness),
}
}
fn stale_daemon_restart_commands() -> Vec<String> {
let mut commands = vec![
"cargo build --release".to_string(),
"cp target/release/batty ~/.cargo/bin/batty".to_string(),
];
#[cfg(target_os = "macos")]
commands.push("codesign --force --sign - ~/.cargo/bin/batty".to_string());
commands.push("batty stop".to_string());
commands.push("batty start".to_string());
commands
}
fn release_batty_path(project_root: &Path) -> PathBuf {
project_root.join("target").join("release").join("batty")
}
fn merge_lock_path(project_root: &Path) -> PathBuf {
project_root.join(".batty").join("merge.lock")
}
fn installed_batty_path() -> Result<PathBuf> {
let home =
std::env::var_os("HOME").context("HOME is not set; cannot resolve ~/.cargo/bin/batty")?;
Ok(PathBuf::from(home).join(".cargo").join("bin").join("batty"))
}
pub(crate) fn rebuild_reinstall_and_resign_batty(project_root: &Path) -> Result<()> {
run_command(project_root, "cargo", &["build", "--release"])?;
let source = project_root.join("target").join("release").join("batty");
let destination = installed_batty_path()?;
if let Some(parent) = destination.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
}
std::fs::copy(&source, &destination).with_context(|| {
format!(
"failed to copy rebuilt binary from {} to {}",
source.display(),
destination.display()
)
})?;
#[cfg(target_os = "macos")]
run_command(
project_root,
"codesign",
&[
"--force",
"--sign",
"-",
destination.to_string_lossy().as_ref(),
],
)?;
Ok(())
}
fn run_command(project_root: &Path, program: &str, args: &[&str]) -> Result<()> {
let output = std::process::Command::new(program)
.args(args)
.current_dir(project_root)
.output()
.with_context(|| format!("failed to run {program} {}", args.join(" ")))?;
if output.status.success() {
return Ok(());
}
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
bail!(
"command failed: {program} {}\nstdout: {}\nstderr: {}",
args.join(" "),
stdout,
stderr
)
}
fn load_watchdog_state(project_root: &Path) -> Result<PersistedWatchdogState> {
let path = watchdog_state_path(project_root);
if !path.exists() {
return Ok(PersistedWatchdogState::default());
}
let content = std::fs::read_to_string(&path)
.with_context(|| format!("failed to read {}", path.display()))?;
serde_json::from_str(&content).with_context(|| format!("failed to parse {}", path.display()))
}
fn save_watchdog_state(project_root: &Path, state: &PersistedWatchdogState) -> Result<()> {
let path = watchdog_state_path(project_root);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
}
let content =
serde_json::to_string_pretty(state).context("failed to serialize watchdog state")?;
std::fs::write(&path, content).with_context(|| format!("failed to write {}", path.display()))
}
fn spawn_detached_process(
project_root: &Path,
args: &[String],
pid_path: &Path,
process_name: &str,
) -> Result<u32> {
use std::process::{Command, Stdio};
let log_path = daemon_log_path(project_root);
if let Some(parent) = log_path.parent() {
std::fs::create_dir_all(parent)?;
}
let log_file = open_log_for_append(&log_path)?;
let log_err = log_file
.try_clone()
.context("failed to clone log file handle")?;
let exe = std::env::current_exe().context("failed to resolve current executable")?;
let mut cmd = Command::new(exe);
cmd.args(args)
.stdin(Stdio::null())
.stdout(log_file)
.stderr(log_err);
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
cmd.process_group(0);
}
let mut child = cmd.spawn().context("failed to spawn daemon process")?;
let pid = child.id();
std::thread::sleep(std::time::Duration::from_millis(500));
match child.try_wait() {
Ok(Some(status)) => {
let _ = std::fs::remove_file(pid_path);
let tail = std::fs::read_to_string(&log_path).ok().and_then(|s| {
let lines: Vec<&str> = s.lines().collect();
let start = lines.len().saturating_sub(5);
let tail = lines[start..].join("\n");
if tail.trim().is_empty() {
None
} else {
Some(tail)
}
});
match tail {
Some(detail) => bail!(
"{process_name} process exited immediately with {status}\n\n\
{detail}\n\n\
see full log: {log}",
log = log_path.display(),
),
None => bail!(
"{process_name} process exited immediately with {status}; \
see {log} for details",
log = log_path.display(),
),
}
}
Ok(None) => {} Err(e) => {
warn!(pid, error = %e, "failed to check daemon process status");
}
}
std::fs::write(pid_path, pid.to_string())
.with_context(|| format!("failed to write PID file: {}", pid_path.display()))?;
info!(pid, log = %log_path.display(), process = process_name, "background process spawned");
Ok(pid)
}
fn spawn_watchdog(project_root: &Path, resume: bool) -> Result<u32> {
let root_str = project_root
.canonicalize()
.unwrap_or_else(|_| project_root.to_path_buf())
.to_string_lossy()
.to_string();
let args = watchdog_spawn_args(&root_str, resume);
spawn_detached_process(
project_root,
&args,
&watchdog_pid_path(project_root),
"watchdog",
)
}
fn ensure_no_concurrent_batty_process(project_root: &Path) -> Result<()> {
match super::process_tree::concurrent_batty_process(std::process::id(), project_root) {
Ok(Some(process)) => {
bail!(
"daemon startup pre-flight failed: another batty process is already running \
for this project (pid {}, command '{}')",
process.pid,
process.command
);
}
Ok(None) => Ok(()),
Err(error) => {
warn!(
error = %error,
"skipping concurrent batty process preflight; process ancestry unavailable"
);
Ok(())
}
}
}
fn spawn_daemon_child(project_root: &Path, resume: bool) -> Result<std::process::Child> {
use std::process::Command;
let exe = std::env::current_exe().context("failed to resolve current executable")?;
let root_str = project_root
.canonicalize()
.unwrap_or_else(|_| project_root.to_path_buf())
.to_string_lossy()
.to_string();
let mut cmd = Command::new(exe);
cmd.args(daemon_spawn_args(&root_str, resume));
cmd.spawn().context("failed to spawn daemon child")
}
#[cfg(unix)]
extern "C" fn handle_watchdog_shutdown_signal(_signal: libc::c_int) {
WATCHDOG_SHUTDOWN_REQUESTED.store(true, Ordering::SeqCst);
}
#[cfg(unix)]
fn install_watchdog_signal_handlers() -> Result<()> {
unsafe {
libc::signal(
libc::SIGTERM,
handle_watchdog_shutdown_signal as *const () as libc::sighandler_t,
);
libc::signal(
libc::SIGINT,
handle_watchdog_shutdown_signal as *const () as libc::sighandler_t,
);
libc::signal(
libc::SIGHUP,
handle_watchdog_shutdown_signal as *const () as libc::sighandler_t,
);
}
WATCHDOG_SHUTDOWN_REQUESTED.store(false, Ordering::SeqCst);
Ok(())
}
#[cfg(not(unix))]
fn install_watchdog_signal_handlers() -> Result<()> {
Ok(())
}
#[cfg(unix)]
fn watchdog_shutdown_requested() -> bool {
WATCHDOG_SHUTDOWN_REQUESTED.load(Ordering::SeqCst)
}
#[cfg(not(unix))]
fn watchdog_shutdown_requested() -> bool {
false
}
fn event_count(path: &Path) -> usize {
events::read_events(path)
.map(|events| events.len())
.unwrap_or(0)
}
fn read_daemon_exit_observation(
project_root: &Path,
event_cursor: usize,
) -> Option<DaemonExitObservation> {
let events_path = super::team_events_path(project_root);
let events = events::read_events(&events_path).ok()?;
events
.into_iter()
.skip(event_cursor)
.rev()
.find_map(|event| {
if event.event == "daemon_exited" || event.event == "daemon_stopped" {
Some(DaemonExitObservation {
reason: event.reason?,
exit_category: event
.exit_category
.unwrap_or_else(|| DAEMON_EXIT_CATEGORY_UNKNOWN.to_string()),
})
} else if event.event == "daemon_panic" {
Some(DaemonExitObservation {
reason: event.reason.unwrap_or_else(|| "unknown panic".to_string()),
exit_category: DAEMON_EXIT_CATEGORY_UNKNOWN.to_string(),
})
} else {
None
}
})
}
fn observe_daemon_exit(
project_root: &Path,
event_cursor: usize,
exit_status: std::process::ExitStatus,
) -> DaemonExitObservation {
read_daemon_exit_observation(project_root, event_cursor).unwrap_or_else(|| {
let reason = if let Some(code) = exit_status.code() {
format!("daemon exited with status {code}")
} else {
"daemon exited from signal".to_string()
};
DaemonExitObservation {
exit_category: classify_daemon_exit_reason(&reason).to_string(),
reason,
}
})
}
fn record_watchdog_crash(
project_root: &Path,
state: &mut PersistedWatchdogState,
observation: DaemonExitObservation,
) -> Result<Option<u64>> {
let now = super::now_unix();
state.restart_count += 1;
state.last_exit_reason = Some(observation.reason.clone());
state.last_exit_category = Some(observation.exit_category.clone());
state.child_pid = None;
state
.crash_timestamps
.retain(|ts| now.saturating_sub(*ts) < WATCHDOG_CIRCUIT_BREAKER_WINDOW_SECS);
state.crash_timestamps.push(now);
if observation.exit_category == DAEMON_EXIT_CATEGORY_UNRECOVERABLE {
state.circuit_breaker_tripped = true;
state.current_backoff_secs = None;
save_watchdog_state(project_root, state)?;
return Ok(None);
}
if state.crash_timestamps.len() >= WATCHDOG_CIRCUIT_BREAKER_THRESHOLD {
state.circuit_breaker_tripped = true;
state.current_backoff_secs = None;
save_watchdog_state(project_root, state)?;
return Ok(None);
}
let exponent = state.crash_timestamps.len().saturating_sub(1) as u32;
let backoff_secs = (WATCHDOG_INITIAL_BACKOFF_SECS
.saturating_mul(2u64.saturating_pow(exponent)))
.min(WATCHDOG_MAX_BACKOFF_SECS);
state.current_backoff_secs = Some(backoff_secs);
save_watchdog_state(project_root, state)?;
Ok(Some(backoff_secs))
}
fn clear_watchdog_child_pid(project_root: &Path, state: &mut PersistedWatchdogState) -> Result<()> {
state.child_pid = None;
let _ = std::fs::remove_file(daemon_child_pid_path(project_root));
save_watchdog_state(project_root, state)
}
fn terminate_daemon_child(child: &mut std::process::Child) {
#[cfg(unix)]
{
let _ = send_unix_signal(child.id(), libc::SIGTERM);
}
let deadline = std::time::Instant::now() + DAEMON_SHUTDOWN_GRACE_PERIOD;
loop {
match child.try_wait() {
Ok(Some(_)) => return,
Ok(None) if std::time::Instant::now() < deadline => {
std::thread::sleep(DAEMON_SHUTDOWN_POLL_INTERVAL);
}
Ok(None) | Err(_) => {
#[cfg(unix)]
{
let _ = send_unix_signal(child.id(), libc::SIGKILL);
}
let _ = child.wait();
return;
}
}
}
}
fn read_daemon_pid(project_root: &Path) -> Option<u32> {
let pid_path = watchdog_pid_path(project_root);
let pid_str = std::fs::read_to_string(pid_path).ok()?;
pid_str.trim().parse::<u32>().ok()
}
#[cfg(unix)]
fn send_unix_signal(pid: u32, signal: libc::c_int) -> bool {
let status = unsafe { libc::kill(pid as libc::pid_t, signal) };
if status == 0 {
true
} else {
let error = std::io::Error::last_os_error();
warn!(pid, signal, error = %error, "failed to signal daemon");
false
}
}
#[cfg(not(unix))]
fn send_unix_signal(_pid: u32, _signal: i32) -> bool {
false
}
#[cfg(unix)]
fn daemon_process_exists(pid: u32) -> bool {
let status = unsafe { libc::kill(pid as libc::pid_t, 0) };
if status == 0 {
!process_is_zombie(pid).unwrap_or(false)
} else {
!matches!(
std::io::Error::last_os_error().raw_os_error(),
Some(libc::ESRCH)
)
}
}
#[cfg(unix)]
fn process_is_zombie(pid: u32) -> Option<bool> {
let output = std::process::Command::new("ps")
.args(["-o", "stat=", "-p", &pid.to_string()])
.output()
.ok()?;
if !output.status.success() {
return None;
}
let stat = String::from_utf8_lossy(&output.stdout);
Some(stat.trim_start().starts_with('Z'))
}
#[cfg(not(unix))]
fn daemon_process_exists(_pid: u32) -> bool {
false
}
pub(crate) fn process_exists(pid: u32) -> bool {
daemon_process_exists(pid)
}
fn wait_for_graceful_daemon_shutdown(
project_root: &Path,
pid: u32,
previous_saved_at: Option<u64>,
timeout: Duration,
) -> bool {
let deadline = std::time::Instant::now() + timeout;
loop {
let clean_snapshot = daemon_state_indicates_clean_shutdown(project_root, previous_saved_at);
if clean_snapshot {
let _ = std::fs::remove_file(watchdog_pid_path(project_root));
return true;
}
let running = daemon_process_exists(pid);
if !running {
let _ = std::fs::remove_file(watchdog_pid_path(project_root));
return false;
}
if std::time::Instant::now() >= deadline {
return false;
}
std::thread::sleep(DAEMON_SHUTDOWN_POLL_INTERVAL);
}
}
pub(super) fn request_graceful_daemon_shutdown(project_root: &Path, timeout: Duration) -> bool {
let Some(pid) = read_daemon_pid(project_root) else {
return true;
};
let previous_saved_at = read_daemon_state_probe(project_root).and_then(|state| state.saved_at);
#[cfg(unix)]
{
if !send_unix_signal(pid, libc::SIGTERM) {
return false;
}
info!(pid, "sent SIGTERM to daemon");
}
#[cfg(not(unix))]
{
warn!(
pid,
"graceful daemon shutdown is not supported on this platform"
);
return false;
}
wait_for_graceful_daemon_shutdown(project_root, pid, previous_saved_at, timeout)
}
pub(super) fn force_kill_daemon(project_root: &Path) {
let Some(pid) = read_daemon_pid(project_root) else {
return;
};
#[cfg(unix)]
{
if send_unix_signal(pid, libc::SIGKILL) {
info!(pid, "sent SIGKILL to daemon");
}
}
#[cfg(not(unix))]
{
warn!(pid, "cannot force-kill daemon on this platform");
}
let _ = std::fs::remove_file(watchdog_pid_path(project_root));
}
pub fn start_team(project_root: &Path, attach: bool) -> Result<String> {
let config_path = team_config_path(project_root);
if !config_path.exists() {
bail!(
"no team config found at {}; run `batty init` first",
config_path.display()
);
}
let team_config = config::TeamConfig::load(&config_path)?;
team_config.validate()?;
team_config.validate_project_refs(project_root)?;
let members = hierarchy::resolve_hierarchy(&team_config)?;
let session = format!("batty-{}", team_config.name);
ensure_no_concurrent_batty_process(project_root)?;
if tmux::session_exists(&session) {
bail!("session '{session}' already exists; use `batty attach` or `batty stop` first");
}
layout::build_layout(
&session,
&members,
&team_config.layout,
project_root,
team_config.workflow_mode,
team_config.orchestrator_enabled(),
team_config.orchestrator_position,
)?;
let inboxes = inbox::inboxes_root(project_root);
for member in &members {
inbox::init_inbox(&inboxes, &member.name)?;
}
let marker = resume_marker_path(project_root);
let resume = marker.exists() || should_resume_from_daemon_state(project_root);
if resume {
if marker.exists() {
std::fs::remove_file(&marker).ok();
}
info!("resuming agent sessions from previous run");
}
info!(session = %session, members = members.len(), resume, "team session started");
let pid = spawn_watchdog(project_root, resume)?;
info!(pid, "watchdog process launched");
std::thread::sleep(std::time::Duration::from_secs(2));
if attach {
tmux::attach(&session)?;
}
Ok(session)
}
pub fn run_daemon(project_root: &Path, resume: bool) -> Result<()> {
let config_path = team_config_path(project_root);
if !config_path.exists() {
bail!(
"no team config found at {}; run `batty init` first",
config_path.display()
);
}
let team_config = config::TeamConfig::load(&config_path)?;
team_config.validate_project_refs(project_root)?;
let members = hierarchy::resolve_hierarchy(&team_config)?;
let session = format!("batty-{}", team_config.name);
for _ in 0..30 {
if tmux::session_exists(&session) {
break;
}
std::thread::sleep(std::time::Duration::from_millis(200));
}
if !tmux::session_exists(&session) {
bail!("tmux session '{session}' not found — did `batty start` create it?");
}
let mut pane_map = std::collections::HashMap::new();
for member in &members {
if let Some(pane_id) = find_pane_for_member(&session, &member.name) {
pane_map.insert(member.name.clone(), pane_id);
}
}
let daemon_config = daemon::DaemonConfig {
project_root: project_root.to_path_buf(),
team_config,
session,
members,
pane_map,
};
let events_path = project_root
.join(".batty")
.join("team_config")
.join("events.jsonl");
let mut d = daemon::TeamDaemon::new(daemon_config)?;
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| d.run(resume)));
match result {
Ok(Ok(())) => Ok(()),
Ok(Err(e)) => {
error!(error = %e, "daemon exited with error");
eprintln!("daemon exited with error: {e:#}");
if let Ok(mut sink) = events::EventSink::new(&events_path) {
let reason = format!("error: {e:#}");
let exit_category = classify_daemon_exit_reason(&reason);
let _ = sink.emit(events::TeamEvent::daemon_exited(&reason, 0, exit_category));
let _ = sink.emit(events::TeamEvent::daemon_stopped_with_reason_and_category(
&reason,
0,
Some(exit_category),
));
}
Err(e)
}
Err(panic_payload) => {
let reason = match panic_payload.downcast_ref::<&str>() {
Some(s) => s.to_string(),
None => match panic_payload.downcast_ref::<String>() {
Some(s) => s.clone(),
None => "unknown panic".to_string(),
},
};
error!(reason = %reason, "daemon panicked");
eprintln!("daemon panicked: {reason}");
if let Ok(mut sink) = events::EventSink::new(&events_path) {
let _ = sink.emit(events::TeamEvent::daemon_exited(
&format!("panic: {reason}"),
0,
DAEMON_EXIT_CATEGORY_UNKNOWN,
));
let _ = sink.emit(events::TeamEvent::daemon_panic(&reason));
}
std::panic::resume_unwind(panic_payload);
}
}
}
pub fn run_watchdog(project_root: &Path, resume: bool) -> Result<()> {
install_watchdog_signal_handlers()?;
let mut state = load_watchdog_state(project_root).unwrap_or_default();
state.circuit_breaker_tripped = false;
state.current_backoff_secs = None;
state.last_exit_reason = None;
state.last_exit_category = None;
state.child_pid = None;
save_watchdog_state(project_root, &state)?;
let mut resume_on_launch = resume;
loop {
if watchdog_shutdown_requested() {
let _ = std::fs::remove_file(watchdog_pid_path(project_root));
let _ = std::fs::remove_file(daemon_child_pid_path(project_root));
state.child_pid = None;
state.current_backoff_secs = None;
save_watchdog_state(project_root, &state)?;
return Ok(());
}
let event_cursor = event_count(&super::team_events_path(project_root));
let mut child = spawn_daemon_child(project_root, resume_on_launch)?;
resume_on_launch = true;
state.child_pid = Some(child.id());
state.current_backoff_secs = None;
save_watchdog_state(project_root, &state)?;
std::fs::write(daemon_child_pid_path(project_root), child.id().to_string()).with_context(
|| {
format!(
"failed to write child PID file: {}",
daemon_child_pid_path(project_root).display()
)
},
)?;
loop {
if watchdog_shutdown_requested() {
terminate_daemon_child(&mut child);
let _ = std::fs::remove_file(watchdog_pid_path(project_root));
clear_watchdog_child_pid(project_root, &mut state)?;
return Ok(());
}
match child.try_wait() {
Ok(Some(exit_status)) => {
clear_watchdog_child_pid(project_root, &mut state)?;
let observation = observe_daemon_exit(project_root, event_cursor, exit_status);
if let Some(backoff_secs) =
record_watchdog_crash(project_root, &mut state, observation.clone())?
{
warn!(
backoff_secs,
reason = %observation.reason,
exit_category = %observation.exit_category,
"daemon crashed; watchdog restarting with backoff"
);
std::thread::sleep(Duration::from_secs(backoff_secs));
break;
}
warn!(
reason = %observation.reason,
exit_category = %observation.exit_category,
threshold = WATCHDOG_CIRCUIT_BREAKER_THRESHOLD,
window_secs = WATCHDOG_CIRCUIT_BREAKER_WINDOW_SECS,
"watchdog circuit breaker tripped; daemon will not be restarted"
);
let _ = std::fs::remove_file(watchdog_pid_path(project_root));
return Ok(());
}
Ok(None) => std::thread::sleep(WATCHDOG_POLL_INTERVAL),
Err(error) => {
clear_watchdog_child_pid(project_root, &mut state)?;
let reason = format!("failed to poll daemon child: {error}");
let observation = DaemonExitObservation {
exit_category: classify_daemon_exit_reason(&reason).to_string(),
reason: reason.clone(),
};
if let Some(backoff_secs) =
record_watchdog_crash(project_root, &mut state, observation)?
{
warn!(backoff_secs, reason = %reason, "watchdog poll failed; retrying daemon launch");
std::thread::sleep(Duration::from_secs(backoff_secs));
break;
}
warn!(
reason = %reason,
threshold = WATCHDOG_CIRCUIT_BREAKER_THRESHOLD,
window_secs = WATCHDOG_CIRCUIT_BREAKER_WINDOW_SECS,
"watchdog circuit breaker tripped after daemon poll failures"
);
let _ = std::fs::remove_file(watchdog_pid_path(project_root));
return Ok(());
}
}
}
}
}
fn find_pane_for_member(session: &str, member_name: &str) -> Option<String> {
let output = crate::tmux::run_tmux_with_timeout(
[
"list-panes",
"-t",
session,
"-F",
"#{pane_id} #{@batty_role}",
],
"list-panes @batty_role",
Some(session),
)
.ok()?;
if !output.status.success() {
return None;
}
let stdout = String::from_utf8_lossy(&output.stdout);
for line in stdout.lines() {
let parts: Vec<&str> = line.splitn(2, ' ').collect();
if parts.len() == 2 && parts[1] == member_name {
return Some(parts[0].to_string());
}
}
None
}
pub(super) fn resume_marker_path(project_root: &Path) -> PathBuf {
project_root.join(".batty").join("resume")
}
#[derive(Debug, Deserialize)]
struct DaemonStateResumeProbe {
#[serde(default)]
clean_shutdown: bool,
#[serde(default)]
saved_at: Option<u64>,
}
fn read_daemon_state_probe(project_root: &Path) -> Option<DaemonStateResumeProbe> {
let path = daemon_state_path(project_root);
let content = std::fs::read_to_string(&path).ok()?;
match serde_json::from_str::<DaemonStateResumeProbe>(&content) {
Ok(state) => Some(state),
Err(error) => {
warn!(
path = %path.display(),
error = %error,
"failed to parse daemon state while probing for resume"
);
None
}
}
}
fn daemon_state_indicates_clean_shutdown(
project_root: &Path,
previous_saved_at: Option<u64>,
) -> bool {
let Some(state) = read_daemon_state_probe(project_root) else {
return false;
};
state.clean_shutdown
&& match (state.saved_at, previous_saved_at) {
(Some(saved_at), Some(previous_saved_at)) => saved_at > previous_saved_at,
(Some(_), None) => true,
(None, Some(_)) => false,
(None, None) => true,
}
}
fn should_resume_from_daemon_state(project_root: &Path) -> bool {
read_daemon_state_probe(project_root)
.map(|state| !state.clean_shutdown)
.unwrap_or(false)
}
#[cfg(test)]
mod tests {
use super::*;
use serial_test::serial;
use std::process::Command;
fn init_repo(dir: &Path) {
run_git(dir, &["init", "-q", "-b", "main"]);
run_git(dir, &["config", "user.email", "test@example.com"]);
run_git(dir, &["config", "user.name", "Test"]);
}
fn run_git(dir: &Path, args: &[&str]) {
let status = Command::new("git")
.args(args)
.current_dir(dir)
.status()
.expect("git binary");
assert!(status.success(), "git {args:?} failed in {dir:?}");
}
fn commit_file_with_time(dir: &Path, rel: &str, content: &str, unix_ts: i64) {
let path = dir.join(rel);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).unwrap();
}
std::fs::write(&path, content).unwrap();
run_git(dir, &["add", rel]);
let date = format!("{unix_ts} +0000");
let status = Command::new("git")
.args(["commit", "-q", "-m", &format!("commit {rel}")])
.env("GIT_AUTHOR_DATE", &date)
.env("GIT_COMMITTER_DATE", &date)
.current_dir(dir)
.status()
.expect("git commit");
assert!(status.success());
}
fn fake_binary(dir: &Path, unix_ts: i64) -> PathBuf {
let binary = dir.join(".batty").join("test-batty-bin");
std::fs::create_dir_all(binary.parent().unwrap()).unwrap();
std::fs::write(&binary, "fake batty binary").unwrap();
filetime::set_file_mtime(&binary, filetime::FileTime::from_unix_time(unix_ts, 0)).unwrap();
binary
}
fn fake_release_binary(dir: &Path) -> PathBuf {
let binary = release_batty_path(dir);
std::fs::create_dir_all(binary.parent().unwrap()).unwrap();
std::fs::write(&binary, "fake release batty binary").unwrap();
binary
}
#[test]
fn daemon_state_probe_requests_resume_after_unclean_shutdown() {
let tmp = tempfile::tempdir().unwrap();
let path = daemon_state_path(tmp.path());
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
std::fs::write(&path, r#"{"clean_shutdown":false}"#).unwrap();
assert!(should_resume_from_daemon_state(tmp.path()));
}
#[test]
fn daemon_state_probe_ignores_clean_shutdown() {
let tmp = tempfile::tempdir().unwrap();
let path = daemon_state_path(tmp.path());
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
std::fs::write(&path, r#"{"clean_shutdown":true}"#).unwrap();
assert!(!should_resume_from_daemon_state(tmp.path()));
}
#[test]
fn daemon_restart_if_stale_fresh_binary_is_noop() {
let tmp = tempfile::tempdir().unwrap();
let repo = tmp.path();
init_repo(repo);
commit_file_with_time(repo, "src/lib.rs", "pub fn fresh() {}\n", 1_700_000_000);
let binary = fake_binary(repo, 1_700_000_000);
let report = restart_daemon_if_stale_with_binary_path(repo, &binary, false).unwrap();
assert_eq!(report.status, StaleDaemonRestartStatus::FreshNoop);
assert!(report.commands.is_empty());
assert!(report.message.contains("fresh"));
}
#[test]
fn daemon_restart_if_stale_clean_root_dry_run_reports_safe_plan() {
let tmp = tempfile::tempdir().unwrap();
let repo = tmp.path();
init_repo(repo);
commit_file_with_time(repo, "src/lib.rs", "pub fn one() {}\n", 1_700_000_000);
commit_file_with_time(repo, "src/main.rs", "pub fn two() {}\n", 1_700_001_000);
let binary = fake_binary(repo, 1_700_000_000);
fake_release_binary(repo);
let report = restart_daemon_if_stale_with_binary_path(repo, &binary, true).unwrap();
assert_eq!(report.status, StaleDaemonRestartStatus::DryRunReady);
assert!(report.dirty_paths.is_empty());
assert!(
report
.commands
.iter()
.any(|command| command == "cargo build --release")
);
assert!(
report
.commands
.iter()
.any(|command| command == "batty start")
);
assert!(report.render().contains("safe restart path is ready"));
}
#[test]
fn daemon_restart_if_stale_refuses_dirty_source_paths() {
let tmp = tempfile::tempdir().unwrap();
let repo = tmp.path();
init_repo(repo);
commit_file_with_time(repo, "src/lib.rs", "pub fn one() {}\n", 1_700_000_000);
commit_file_with_time(repo, "src/main.rs", "pub fn two() {}\n", 1_700_001_000);
std::fs::write(repo.join("src").join("dirty.rs"), "pub fn dirty() {}\n").unwrap();
let binary = fake_binary(repo, 1_700_000_000);
fake_release_binary(repo);
let report = restart_daemon_if_stale_with_binary_path(repo, &binary, true).unwrap();
assert_eq!(report.status, StaleDaemonRestartStatus::RefusedDirty);
assert_eq!(report.dirty_paths, vec!["src/dirty.rs"]);
assert!(report.render().contains("Dirty source paths: src/dirty.rs"));
assert!(report.render().contains(STALE_MANUAL_RECOVERY_COMMAND));
}
#[test]
fn daemon_restart_if_stale_allows_runtime_only_dirty_paths() {
let tmp = tempfile::tempdir().unwrap();
let repo = tmp.path();
init_repo(repo);
commit_file_with_time(repo, "src/lib.rs", "pub fn one() {}\n", 1_700_000_000);
commit_file_with_time(repo, "src/main.rs", "pub fn two() {}\n", 1_700_001_000);
let telemetry = repo.join(".batty").join("telemetry.db");
std::fs::create_dir_all(telemetry.parent().unwrap()).unwrap();
std::fs::write(&telemetry, "runtime noise\n").unwrap();
let binary = fake_binary(repo, 1_700_000_000);
fake_release_binary(repo);
let report = restart_daemon_if_stale_with_binary_path(repo, &binary, true).unwrap();
assert_eq!(report.status, StaleDaemonRestartStatus::DryRunReady);
assert!(report.dirty_paths.is_empty());
assert!(
report
.render()
.contains("Runtime-only dirty paths are tolerated")
);
}
#[test]
fn daemon_restart_if_stale_refuses_active_merge_lock() {
let tmp = tempfile::tempdir().unwrap();
let repo = tmp.path();
init_repo(repo);
commit_file_with_time(repo, "src/lib.rs", "pub fn one() {}\n", 1_700_000_000);
commit_file_with_time(repo, "src/main.rs", "pub fn two() {}\n", 1_700_001_000);
let binary = fake_binary(repo, 1_700_000_000);
fake_release_binary(repo);
let lock = repo.join(".batty").join("merge.lock");
std::fs::create_dir_all(lock.parent().unwrap()).unwrap();
std::fs::write(lock, "active merge\n").unwrap();
let report = restart_daemon_if_stale_with_binary_path(repo, &binary, true).unwrap();
assert_eq!(report.status, StaleDaemonRestartStatus::RefusedActiveMerge);
assert!(report.render().contains("merge is active"));
}
#[test]
fn daemon_restart_if_stale_refuses_missing_release_binary() {
let tmp = tempfile::tempdir().unwrap();
let repo = tmp.path();
init_repo(repo);
commit_file_with_time(repo, "src/lib.rs", "pub fn one() {}\n", 1_700_000_000);
commit_file_with_time(repo, "src/main.rs", "pub fn two() {}\n", 1_700_001_000);
let binary = fake_binary(repo, 1_700_000_000);
let report = restart_daemon_if_stale_with_binary_path(repo, &binary, true).unwrap();
assert_eq!(
report.status,
StaleDaemonRestartStatus::RefusedMissingReleaseBinary
);
assert!(report.render().contains("release binary is missing"));
}
#[cfg(unix)]
fn write_daemon_script(script_path: &Path, body: &str) {
std::fs::write(script_path, body).unwrap();
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(script_path, std::fs::Permissions::from_mode(0o755)).unwrap();
}
#[cfg(unix)]
#[test]
#[serial]
fn graceful_daemon_shutdown_waits_for_clean_snapshot() {
let tmp = tempfile::tempdir().unwrap();
let state_path = daemon_state_path(tmp.path());
let state_dir = state_path.parent().unwrap();
std::fs::create_dir_all(state_dir).unwrap();
std::fs::write(&state_path, r#"{"clean_shutdown":false,"saved_at":1}"#).unwrap();
let state_path_for_thread = state_path.clone();
let state_dir_for_thread = state_dir.to_path_buf();
let writer = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(200));
std::fs::create_dir_all(&state_dir_for_thread).unwrap();
std::fs::write(
&state_path_for_thread,
r#"{"clean_shutdown":true,"saved_at":2}"#,
)
.unwrap();
});
assert!(wait_for_graceful_daemon_shutdown(
tmp.path(),
std::process::id(),
Some(1),
Duration::from_secs(2)
));
writer.join().unwrap();
assert!(daemon_state_indicates_clean_shutdown(tmp.path(), Some(1)));
}
#[cfg(unix)]
#[test]
#[serial]
fn graceful_daemon_shutdown_times_out_before_force_kill_fallback() {
let tmp = tempfile::tempdir().unwrap();
let script_path = tmp.path().join("stubborn-daemon.sh");
write_daemon_script(
&script_path,
"#!/bin/sh\ntrap '' TERM\nwhile :; do :; done\n",
);
let mut child = std::process::Command::new(&script_path).spawn().unwrap();
std::fs::create_dir_all(tmp.path().join(".batty")).unwrap();
std::fs::write(watchdog_pid_path(tmp.path()), child.id().to_string()).unwrap();
std::thread::sleep(Duration::from_millis(200));
assert!(!request_graceful_daemon_shutdown(
tmp.path(),
Duration::from_millis(300)
));
assert!(daemon_process_exists(child.id()));
force_kill_daemon(tmp.path());
let _ = child.wait().unwrap();
assert!(!watchdog_pid_path(tmp.path()).exists());
}
#[test]
fn test_rotate_log_shifts_files() {
let tmp = tempfile::tempdir().unwrap();
let log_path = daemon_log_path(tmp.path());
std::fs::create_dir_all(log_path.parent().unwrap()).unwrap();
std::fs::write(&log_path, b"current").unwrap();
std::fs::write(rotated_log_path(&log_path, 1), b"older-1").unwrap();
std::fs::write(rotated_log_path(&log_path, 2), b"older-2").unwrap();
std::fs::OpenOptions::new()
.write(true)
.open(&log_path)
.unwrap()
.set_len(LOG_ROTATION_BYTES + 1)
.unwrap();
rotate_log_if_needed(&log_path).unwrap();
assert!(!log_path.exists());
assert_eq!(
std::fs::read(rotated_log_path(&log_path, 1)).unwrap().len() as u64,
LOG_ROTATION_BYTES + 1
);
assert_eq!(
std::fs::read_to_string(rotated_log_path(&log_path, 2)).unwrap(),
"older-1"
);
assert_eq!(
std::fs::read_to_string(rotated_log_path(&log_path, 3)).unwrap(),
"older-2"
);
}
#[test]
fn test_rotate_log_keeps_max_3() {
let tmp = tempfile::tempdir().unwrap();
let log_path = crate::team::orchestrator_log_path(tmp.path());
std::fs::create_dir_all(log_path.parent().unwrap()).unwrap();
std::fs::write(&log_path, b"current").unwrap();
std::fs::write(rotated_log_path(&log_path, 1), b"older-1").unwrap();
std::fs::write(rotated_log_path(&log_path, 2), b"older-2").unwrap();
std::fs::write(rotated_log_path(&log_path, 3), b"older-3").unwrap();
std::fs::OpenOptions::new()
.write(true)
.open(&log_path)
.unwrap()
.set_len(LOG_ROTATION_BYTES + 1)
.unwrap();
rotate_log_if_needed(&log_path).unwrap();
assert_eq!(
std::fs::read(rotated_log_path(&log_path, 1)).unwrap().len() as u64,
LOG_ROTATION_BYTES + 1
);
assert_eq!(
std::fs::read_to_string(rotated_log_path(&log_path, 2)).unwrap(),
"older-1"
);
assert_eq!(
std::fs::read_to_string(rotated_log_path(&log_path, 3)).unwrap(),
"older-2"
);
assert!(!rotated_log_path(&log_path, 4).exists());
}
#[test]
fn test_rotate_log_noop_under_threshold() {
let tmp = tempfile::tempdir().unwrap();
let log_path = daemon_log_path(tmp.path());
std::fs::create_dir_all(log_path.parent().unwrap()).unwrap();
std::fs::write(&log_path, b"small-log").unwrap();
rotate_log_if_needed(&log_path).unwrap();
assert_eq!(std::fs::read_to_string(&log_path).unwrap(), "small-log");
assert!(!rotated_log_path(&log_path, 1).exists());
}
#[test]
fn test_daemon_log_append_mode() {
let tmp = tempfile::tempdir().unwrap();
let log_path = daemon_log_path(tmp.path());
{
let mut file = open_log_for_append(&log_path).unwrap();
use std::io::Write;
writeln!(file, "first").unwrap();
}
{
let mut file = open_log_for_append(&log_path).unwrap();
use std::io::Write;
writeln!(file, "second").unwrap();
}
assert_eq!(
std::fs::read_to_string(&log_path).unwrap(),
"first\nsecond\n"
);
}
#[test]
fn daemon_spawn_args_include_verbose_and_resume() {
assert_eq!(
daemon_spawn_args("/tmp/project", false),
vec![
"-v".to_string(),
"daemon".to_string(),
"--project-root".to_string(),
"/tmp/project".to_string()
]
);
assert_eq!(
daemon_spawn_args("/tmp/project", true),
vec![
"-v".to_string(),
"daemon".to_string(),
"--project-root".to_string(),
"/tmp/project".to_string(),
"--resume".to_string()
]
);
}
#[test]
fn watchdog_spawn_args_include_verbose_and_resume() {
assert_eq!(
watchdog_spawn_args("/tmp/project", false),
vec![
"-v".to_string(),
"watchdog".to_string(),
"--project-root".to_string(),
"/tmp/project".to_string()
]
);
assert_eq!(
watchdog_spawn_args("/tmp/project", true),
vec![
"-v".to_string(),
"watchdog".to_string(),
"--project-root".to_string(),
"/tmp/project".to_string(),
"--resume".to_string()
]
);
}
#[test]
fn record_watchdog_crash_applies_exponential_backoff_until_circuit_breaker() {
let tmp = tempfile::tempdir().unwrap();
let mut state = PersistedWatchdogState::default();
assert_eq!(
record_watchdog_crash(
tmp.path(),
&mut state,
DaemonExitObservation {
reason: "boom-1".to_string(),
exit_category: DAEMON_EXIT_CATEGORY_UNKNOWN.to_string(),
},
)
.unwrap(),
Some(1)
);
assert_eq!(
record_watchdog_crash(
tmp.path(),
&mut state,
DaemonExitObservation {
reason: "boom-2".to_string(),
exit_category: DAEMON_EXIT_CATEGORY_UNKNOWN.to_string(),
},
)
.unwrap(),
Some(2)
);
assert_eq!(
record_watchdog_crash(
tmp.path(),
&mut state,
DaemonExitObservation {
reason: "boom-3".to_string(),
exit_category: DAEMON_EXIT_CATEGORY_UNKNOWN.to_string(),
},
)
.unwrap(),
Some(4)
);
assert_eq!(
record_watchdog_crash(
tmp.path(),
&mut state,
DaemonExitObservation {
reason: "boom-4".to_string(),
exit_category: DAEMON_EXIT_CATEGORY_UNKNOWN.to_string(),
},
)
.unwrap(),
Some(8)
);
assert_eq!(
record_watchdog_crash(
tmp.path(),
&mut state,
DaemonExitObservation {
reason: "boom-5".to_string(),
exit_category: DAEMON_EXIT_CATEGORY_UNKNOWN.to_string(),
},
)
.unwrap(),
None
);
assert!(state.circuit_breaker_tripped);
assert_eq!(state.restart_count, 5);
}
#[test]
fn record_watchdog_crash_opens_circuit_immediately_for_unrecoverable_exit() {
let tmp = tempfile::tempdir().unwrap();
let mut state = PersistedWatchdogState::default();
assert_eq!(
record_watchdog_crash(
tmp.path(),
&mut state,
DaemonExitObservation {
reason: "tmux server died".to_string(),
exit_category: DAEMON_EXIT_CATEGORY_UNRECOVERABLE.to_string(),
},
)
.unwrap(),
None
);
assert!(state.circuit_breaker_tripped);
assert_eq!(state.restart_count, 1);
assert_eq!(state.last_exit_reason.as_deref(), Some("tmux server died"));
assert_eq!(
state.last_exit_category.as_deref(),
Some(DAEMON_EXIT_CATEGORY_UNRECOVERABLE)
);
}
#[test]
fn read_daemon_exit_observation_prefers_structured_exit_event() {
let tmp = tempfile::tempdir().unwrap();
let events_path = tmp
.path()
.join(".batty")
.join("team_config")
.join("events.jsonl");
std::fs::create_dir_all(events_path.parent().unwrap()).unwrap();
let mut sink = events::EventSink::new(&events_path).unwrap();
sink.emit(events::TeamEvent::daemon_started()).unwrap();
let cursor = event_count(&events_path);
sink.emit(events::TeamEvent::daemon_exited(
"daemon startup pre-flight failed: git worktree is dirty:\nM src/team/task_loop.rs",
0,
DAEMON_EXIT_CATEGORY_UNRECOVERABLE,
))
.unwrap();
let observation = read_daemon_exit_observation(tmp.path(), cursor).unwrap();
assert!(observation.reason.contains("dirty"));
assert_eq!(
observation.exit_category,
DAEMON_EXIT_CATEGORY_UNRECOVERABLE
);
}
}