diff --git a/.gitignore b/.gitignore index 9feab12..17da560 100644 --- a/.gitignore +++ b/.gitignore @@ -2,10 +2,8 @@ /examples/*/state.json **/.#* -# Beads / Dolt files (added by bd init) -.dolt/ -*.db -.beads-credential-key +# Kata local overrides +.kata.local.toml # Codex droppings -**/.codex \ No newline at end of file +**/.codex diff --git a/.kata.toml b/.kata.toml new file mode 100644 index 0000000..f881a00 --- /dev/null +++ b/.kata.toml @@ -0,0 +1,4 @@ +version = 1 + +[project] +name = "devloop" diff --git a/AGENTS.md b/AGENTS.md index 8d2ada4..140e7e2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,8 +6,8 @@ configuration-driven engine for local watch/rebuild/reload workflows without hard-coding knowledge of any one repository. ## Working rules -- Use `bd` for task tracking. Create or update issues before substantial - edits. +- Use `kata` for task tracking. Create or update issues before substantial + edits, and assign active work before implementation. - Prefer stable abstractions over repo-specific shortcuts. Put project-specific behavior behind config or hooks. - Prefer small, focused libraries over bespoke implementations when @@ -82,11 +82,11 @@ without hard-coding knowledge of any one repository. `CHANGELOG.md` as part of the same change before commit. - `devloop` uses semantic versioning. Update versions intentionally to match the scope of delivered changes. -- Record follow-up work in `bd`, not as free-form TODO comments. +- Record follow-up work in `kata`, not as free-form TODO comments. ## Session completion 1. File issues for unfinished work or risks discovered during implementation. 2. Run quality gates if code changed. -3. Update issue status in `bd`. +3. Update issue status in `kata`. 4. Summarize the current state, verification, and next steps. diff --git a/CHANGELOG.md b/CHANGELOG.md index 12188b7..b61480f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ All notable changes to `devloop` will be recorded in this file. ## [Unreleased] +## [0.9.2] - 2026-07-01 + +### Fixed +- Managed processes are now started in their own Unix process groups + and stopped as groups, so child processes spawned by a managed command + do not survive `stop_process`, `restart_process`, or `ctrl-c` + shutdown. +- HTTP readiness and liveness probe attempts are now request-bounded, so + one stuck request cannot stall the probe loop or shutdown path. + ## [0.9.1] - 2026-06-24 ### Changed diff --git a/Cargo.lock b/Cargo.lock index 07a8b25..73e01f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -235,7 +235,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "devloop" -version = "0.9.1" +version = "0.9.2" dependencies = [ "anyhow", "axum", @@ -246,6 +246,7 @@ dependencies = [ "rand", "regex", "reqwest", + "rustix", "serde", "serde_json", "tempfile", diff --git a/Cargo.toml b/Cargo.toml index 34848f3..b555784 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "devloop" -version = "0.9.1" +version = "0.9.2" edition = "2024" [dependencies] @@ -13,6 +13,7 @@ pulldown-cmark = "0.13.0" rand = "0.9.2" regex = "1.11.1" reqwest = { version = "0.12.15", default-features = false, features = ["http2", "json", "rustls-tls"] } +rustix = { version = "1.1.4", features = ["process"] } serde = { version = "1.0.219", features = ["derive"] } serde_json = "1.0.140" tokio = { version = "1.45.1", features = ["io-std", "macros", "process", "rt-multi-thread", "signal", "time"] } diff --git a/PLAN.md b/PLAN.md index 2861c09..846620b 100644 --- a/PLAN.md +++ b/PLAN.md @@ -9,7 +9,7 @@ repository as the first client. ### 1. Project scaffolding - Replace default boilerplate with project-specific docs. - Define architecture boundaries between engine, config, and hooks. -- Capture work in `bd`. +- Capture work in `kata`. ### 2. Engine MVP - Load a config file from a target repository. diff --git a/README.md b/README.md index 2e3a46a..30ae8f6 100644 --- a/README.md +++ b/README.md @@ -304,8 +304,8 @@ which runs `cargo fmt` before each commit. Task tracking: ```bash -bd ready -bd show -bd update --status in_progress -bd close +kata ready --agent +kata show --agent +kata assign --agent +kata close --done --message "" --commit ``` diff --git a/docs/behavior.md b/docs/behavior.md index d4f8d65..4466833 100644 --- a/docs/behavior.md +++ b/docs/behavior.md @@ -99,6 +99,9 @@ Managed processes are long-running child commands. - `start_process` is a no-op if the named process is already running. - `restart_process` stops the child, then starts it again. +- Managed processes are started in their own Unix process group, and + stop/restart/shutdown terminates that group so descendant processes do + not survive the supervisor. - `wait_for_process` waits on the configured readiness probe, not just on successful spawning. - `restart = "always"` restarts a child after any exit unless diff --git a/scripts/ci-smoke.sh b/scripts/ci-smoke.sh index 96dde0c..345a61e 100755 --- a/scripts/ci-smoke.sh +++ b/scripts/ci-smoke.sh @@ -64,6 +64,17 @@ trap dump_log ERR cp -R "${fixture_src}/." "${tmp_dir}/" chmod +x "${tmp_dir}/scripts/read-watched.sh" +smoke_port="$( + python3 - <<'PY' +import socket + +with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + print(sock.getsockname()[1]) +PY +)" +sed -i.bak "s/18081/${smoke_port}/g" "${tmp_dir}/devloop.toml" +rm -f "${tmp_dir}/devloop.toml.bak" state_path="${tmp_dir}/.devloop/state.json" devloop_bin="${repo_root}/target/debug/devloop" @@ -107,7 +118,7 @@ while time.time() < deadline: raise SystemExit("timed out waiting for watcher startup") PY -python3 - "$state_path" "$log_path" "${tmp_dir}/watched.txt" <<'PY' +python3 - "$state_path" "$log_path" "${tmp_dir}/watched.txt" "${smoke_port}" <<'PY' import json import pathlib import sys @@ -116,6 +127,7 @@ import time state_path = pathlib.Path(sys.argv[1]) log_path = pathlib.Path(sys.argv[2]) watched_path = pathlib.Path(sys.argv[3]) +smoke_port = sys.argv[4] deadline = time.time() + 15 next_write = 0.0 while time.time() < deadline: @@ -127,7 +139,7 @@ while time.time() < deadline: data = json.loads(state_path.read_text()) if ( data.get("current_value") == "updated" - and data.get("current_url") == "http://127.0.0.1:18081/updated" + and data.get("current_url") == f"http://127.0.0.1:{smoke_port}/updated" ): if "changed value: updated" in log_path.read_text(): sys.exit(0) diff --git a/src/processes.rs b/src/processes.rs index 2545e5b..503f074 100644 --- a/src/processes.rs +++ b/src/processes.rs @@ -6,10 +6,12 @@ use std::time::{Duration, Instant}; use anyhow::{Context, Result, anyhow}; use regex::Regex; +use rustix::io::Errno; +use rustix::process::{Pid, Signal, kill_process_group}; use tokio::io::{AsyncReadExt, AsyncWriteExt, Stderr, Stdout}; use tokio::process::{Child, Command}; use tokio::sync::Mutex; -use tokio::time::sleep; +use tokio::time::{sleep, timeout}; use tracing::{info, warn}; use crate::browser_reload::{BrowserReloadEnvironment, apply_browser_reload_env}; @@ -40,6 +42,7 @@ pub struct ProcessManager<'a> { struct ManagedProcess { child: Child, + process_group: Pid, } struct CommandContext<'a> { @@ -99,7 +102,7 @@ impl<'a> ProcessManager<'a> { let Some(mut child) = self.children.remove(name) else { return Ok(()); }; - terminate_child(name, &mut child.child).await?; + terminate_child(name, &mut child.child, child.process_group).await?; self.supervisor.on_process_stopped(name); Ok(()) } @@ -232,9 +235,11 @@ impl<'a> ProcessManager<'a> { )?; command.stdout(Stdio::piped()); command.stderr(Stdio::piped()); + command.process_group(0); let mut child = command .spawn() .with_context(|| format!("failed to start process '{name}'"))?; + let process_group = child_process_group(name, &child)?; clear_output_state_keys(&spec.output.rules, state)?; let process_name = name.to_owned(); let source_label = process_output_source_label(name, &spec.command); @@ -271,8 +276,13 @@ impl<'a> ProcessManager<'a> { state.clone(), )); } - self.children - .insert(name.to_owned(), ManagedProcess { child }); + self.children.insert( + name.to_owned(), + ManagedProcess { + child, + process_group, + }, + ); self.supervisor.on_process_started(name); info!("started process {}", name); Ok(()) @@ -910,19 +920,51 @@ fn clear_output_state_keys(rules: &[OutputRule], state: &SessionState) -> Result Ok(()) } -async fn terminate_child(name: &str, child: &mut Child) -> Result<()> { +fn child_process_group(name: &str, child: &Child) -> Result { + let id = child + .id() + .ok_or_else(|| anyhow!("process '{name}' exited before its process group was recorded"))?; + Pid::from_raw(id as i32) + .ok_or_else(|| anyhow!("process '{name}' has invalid process group id {id}")) +} + +async fn terminate_child(name: &str, child: &mut Child, process_group: Pid) -> Result<()> { if child.try_wait()?.is_some() { - info!("process {} already exited", name); + signal_process_group(name, process_group, Signal::KILL)?; + info!("process {} already exited; cleaned up process group", name); return Ok(()); } - child - .kill() - .await - .with_context(|| format!("failed to stop process '{name}'"))?; + + signal_process_group(name, process_group, Signal::TERM)?; + match timeout(Duration::from_secs(2), child.wait()).await { + Ok(result) => { + result.with_context(|| format!("failed to wait for process '{name}' after SIGTERM"))?; + signal_process_group(name, process_group, Signal::KILL)?; + } + Err(_) => { + signal_process_group(name, process_group, Signal::KILL)?; + child + .wait() + .await + .with_context(|| format!("failed to stop process '{name}' after SIGKILL"))?; + } + } info!("stopped process {}", name); Ok(()) } +fn signal_process_group(name: &str, process_group: Pid, signal: Signal) -> Result<()> { + match kill_process_group(process_group, signal) { + Ok(()) | Err(Errno::SRCH) => Ok(()), + Err(error) => Err(anyhow!( + "failed to send {:?} to process group for process '{}': {}", + signal, + name, + error + )), + } +} + async fn wait_for_probe( client: &reqwest::Client, name: &str, @@ -981,7 +1023,16 @@ async fn check_probe( state: &SessionState, ) -> Result<()> { match probe { - ProbeSpec::Http { url, .. } => match client.get(url).send().await { + ProbeSpec::Http { + url, + interval_ms, + timeout_ms, + } => match client + .get(url) + .timeout(probe_attempt_timeout(*interval_ms, *timeout_ms)) + .send() + .await + { Ok(response) if response.status().is_success() => { info!("process {} is healthy at {}", name, url); Ok(()) @@ -1006,6 +1057,11 @@ async fn check_probe( } } +fn probe_attempt_timeout(interval_ms: u64, timeout_ms: u64) -> Duration { + let bounded = interval_ms.saturating_mul(2).max(1000).min(timeout_ms); + Duration::from_millis(bounded) +} + fn timeout_error(name: &str, probe: &ProbeSpec) -> anyhow::Error { match probe { ProbeSpec::Http { url, .. } => { @@ -1020,11 +1076,13 @@ fn timeout_error(name: &str, probe: &ProbeSpec) -> anyhow::Error { #[cfg(test)] mod tests { use super::*; - use crate::config::{OutputExtract, ProbeSpec}; + use crate::config::{OutputConfig, OutputExtract, ProbeSpec}; use crate::test_support::{EnvVarGuard, RustLogGuard}; + use rustix::process::test_kill_process; use serde_json::Value; use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; + use tempfile::tempdir; use tokio::io::AsyncReadExt; use tokio::sync::Mutex; @@ -1036,6 +1094,107 @@ mod tests { std::env::temp_dir().join(format!("devloop-process-state-{unique}.json")) } + fn test_config(root: &Path) -> Config { + Config { + root: root.to_path_buf(), + debounce_ms: 100, + watcher: crate::config::WatcherConfig::default(), + state_file: Some(unique_state_path()), + startup_workflows: vec![], + watch: BTreeMap::new(), + process: BTreeMap::new(), + hook: BTreeMap::new(), + event_server: crate::config::EventServerConfig::default(), + browser_reload_server: crate::config::BrowserReloadServerConfig::default(), + event: BTreeMap::new(), + workflow: BTreeMap::new(), + } + } + + #[cfg(unix)] + async fn wait_for_path(path: &Path) { + let started = Instant::now(); + loop { + if path.exists() { + return; + } + assert!( + started.elapsed() < Duration::from_secs(5), + "timed out waiting for {}", + path.display() + ); + sleep(Duration::from_millis(20)).await; + } + } + + #[cfg(unix)] + async fn assert_process_gone(raw_pid: i32) { + let pid = Pid::from_raw(raw_pid).expect("pid"); + let started = Instant::now(); + loop { + if matches!(test_kill_process(pid), Err(Errno::SRCH)) { + return; + } + assert!( + started.elapsed() < Duration::from_secs(5), + "process {raw_pid} still exists after managed process stop" + ); + sleep(Duration::from_millis(20)).await; + } + } + + #[cfg(unix)] + #[tokio::test] + async fn stop_named_terminates_descendant_processes() { + let dir = tempdir().expect("tempdir"); + let pid_path = dir.path().join("descendant.pid"); + let script_path = dir.path().join("spawn-descendant.sh"); + std::fs::write( + &script_path, + format!( + r#"#!/bin/sh +sh -c 'trap "" TERM; while :; do sleep 1; done' & +echo "$!" > "{}" +wait +"#, + pid_path.display() + ), + ) + .expect("write script"); + + let mut config = test_config(dir.path()); + config.process.insert( + "server".into(), + ProcessSpec { + command: vec!["sh".into(), script_path.display().to_string()], + cwd: Some(dir.path().to_path_buf()), + autostart: false, + readiness: None, + liveness: None, + restart: crate::config::RestartPolicy::Never, + env: BTreeMap::new(), + output: OutputConfig::default(), + }, + ); + let state = SessionState::load(unique_state_path()).expect("state"); + let mut manager = ProcessManager::new(&config); + + manager + .start_named("server", &state) + .await + .expect("start process"); + wait_for_path(&pid_path).await; + let descendant_pid = std::fs::read_to_string(&pid_path) + .expect("read pid") + .trim() + .parse::() + .expect("parse pid"); + + manager.stop_named("server").await.expect("stop process"); + + assert_process_gone(descendant_pid).await; + } + #[test] fn extract_url_token_finds_cloudflare_url() { let rule = CompiledOutputRule { @@ -1053,6 +1212,22 @@ mod tests { assert_eq!(value.as_deref(), Some("https://abc.trycloudflare.com")); } + #[test] + fn probe_attempt_timeout_is_bounded_by_probe_timeout() { + assert_eq!( + probe_attempt_timeout(100, 5000), + Duration::from_millis(1000) + ); + assert_eq!( + probe_attempt_timeout(750, 5000), + Duration::from_millis(1500) + ); + assert_eq!( + probe_attempt_timeout(750, 1200), + Duration::from_millis(1200) + ); + } + #[test] fn format_output_line_prefixes_source() { let rendered = format_output_line( diff --git a/src/state.rs b/src/state.rs index c7b40e4..a9fdc29 100644 --- a/src/state.rs +++ b/src/state.rs @@ -122,14 +122,18 @@ pub(crate) fn render_template_values( #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; + static NEXT_STATE_PATH_ID: AtomicU64 = AtomicU64::new(0); + fn unique_state_path() -> PathBuf { let unique = SystemTime::now() .duration_since(UNIX_EPOCH) .expect("system time") .as_nanos(); - std::env::temp_dir().join(format!("devloop-state-{unique}.json")) + let id = NEXT_STATE_PATH_ID.fetch_add(1, Ordering::Relaxed); + std::env::temp_dir().join(format!("devloop-state-{unique}-{id}.json")) } #[test]