diff --git a/.changes/fix-adopted-worker-exit-wait.md b/.changes/fix-adopted-worker-exit-wait.md new file mode 100644 index 000000000..88ca2cb8d --- /dev/null +++ b/.changes/fix-adopted-worker-exit-wait.md @@ -0,0 +1 @@ +- The daemon no longer wakes up every second for each session it keeps running across a restart or update: it now waits on the operating system's process-exit notification (kqueue on macOS, pidfd on Linux), which cuts idle battery drain. On macOS this also stops a `ps` process being spawned every second for each such session. diff --git a/crates/pa-core/README.md b/crates/pa-core/README.md index 1055520e8..5958ebacf 100644 --- a/crates/pa-core/README.md +++ b/crates/pa-core/README.md @@ -23,7 +23,7 @@ logic of its own). The RLM recursion host seam `rlm.create_session`/`rlm.list_subagents`/`rlm.collect`/ `rlm.delete_subagent` host requests call into, with the roster/collect/ selector-error vocabulary the daemon implements over the supervisor link. -Platform wall (`platform`): process control (signals/process groups), file locking, file permissions, and shell selection - every OS-specific behavior in the engine lives there behind cfg-gated implementations (the durable-write rename primitive is pa-telemetry's `rename_onto`, re-exported as `platform::rename_onto`). Scheduled jobs (`cron`, the `AgentCronJobStore` port of `core/cron-jobs.ts`): file-backed job state under session artifacts (`scheduled-jobs.json` partitions) with cross-process locking, plus the public read-only scan (`cron::store::read_scheduled_jobs_artifact`) the update flow's roster projection and boot re-arm read. Tools (bash, edit, ipython + internal rename/stdout), file mutation queue, truncation and rendering rules, RLM kernel lifecycle (IPython spawn/execute/revive), skills loading, system prompt assembly, compaction, harness refinement, settings/config, package manager (npm/git/local source install/remove/list/update against settings, plus `resolve()`: precedence-ranked session resource resolution over configured packages, settings arrays, auto-discovery, and bundled skills), session manager (persist/resume). origin/main +Platform wall (`platform`): process control (signals/process groups, kernel exit waits), file locking, file permissions, and shell selection - every OS-specific behavior in the engine lives there behind cfg-gated implementations (the durable-write rename primitive is pa-telemetry's `rename_onto`, re-exported as `platform::rename_onto`). Scheduled jobs (`cron`, the `AgentCronJobStore` port of `core/cron-jobs.ts`): file-backed job state under session artifacts (`scheduled-jobs.json` partitions) with cross-process locking, plus the public read-only scan (`cron::store::read_scheduled_jobs_artifact`) the update flow's roster projection and boot re-arm read. Tools (bash, edit, ipython + internal rename/stdout), file mutation queue, truncation and rendering rules, RLM kernel lifecycle (IPython spawn/execute/revive), skills loading, system prompt assembly, compaction, harness refinement, settings/config, package manager (npm/git/local source install/remove/list/update against settings, plus `resolve()`: precedence-ranked session resource resolution over configured packages, settings arrays, auto-discovery, and bundled skills), session manager (persist/resume). origin/main Provider resilience policies in `session_engine`: the shared quick-retry policy (TS `provider-retry.ts`: permanent-kind classification, diff --git a/crates/pa-core/src/platform/process.rs b/crates/pa-core/src/platform/process.rs index 541604066..2fc4684de 100644 --- a/crates/pa-core/src/platform/process.rs +++ b/crates/pa-core/src/platform/process.rs @@ -240,25 +240,37 @@ pub fn pid_exists(pid: u32) -> bool { /// The kernel-held process handle (`pidfd_open`): pins the exact process /// behind the pid, so a signal through it ([`pidfd_signal`]) reaches that -/// process even if the numeric pid is recycled afterwards. `None` when -/// the platform or kernel has no pidfd, or the process is already gone -/// (the caller treats an unobtainable handle as never-signal: a missed -/// stop is recoverable, a wrong one is not). +/// process even if the numeric pid is recycled afterwards. The caller +/// treats an unobtainable handle as never-signal: a missed stop is +/// recoverable, a wrong one is not. +/// +/// # Errors +/// +/// Returns the open failure verbatim: `ESRCH` names a process that is +/// already gone, `Unsupported` a platform with no pidfd arm. #[cfg(all( unix, target_os = "linux", any(target_arch = "x86_64", target_arch = "aarch64") ))] -#[must_use] -pub fn open_pidfd(pid: u32) -> Option { +pub fn open_pidfd(pid: u32) -> std::io::Result { // `SYS_pidfd_open`/`SYS_pidfd_send_signal` share their numbers across // x86_64 and aarch64 (the platforms this workspace ships) — Linux only: // pidfd is a Linux syscall family, and the macOS libc crate carries no // `SYS_pidfd_*` constants for the same cfg to compile against. let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) }; - (fd >= 0).then_some(fd as i32) + if fd < 0 { + Err(std::io::Error::last_os_error()) + } else { + Ok(fd as i32) + } } +/// No pidfd arm on this platform or kernel: the handle is unobtainable. +/// +/// # Errors +/// +/// Always `Unsupported`. #[cfg(all( unix, not(all( @@ -266,13 +278,18 @@ pub fn open_pidfd(pid: u32) -> Option { any(target_arch = "x86_64", target_arch = "aarch64") )) ))] -pub fn open_pidfd(_pid: u32) -> Option { - None +pub fn open_pidfd(_pid: u32) -> std::io::Result { + Err(std::io::ErrorKind::Unsupported.into()) } +/// No pidfd arm on this platform or kernel: the handle is unobtainable. +/// +/// # Errors +/// +/// Always `Unsupported`. #[cfg(not(unix))] -pub fn open_pidfd(_pid: u32) -> Option { - None +pub fn open_pidfd(_pid: u32) -> std::io::Result { + Err(std::io::ErrorKind::Unsupported.into()) } /// Signal through the kernel-held handle (`pidfd_send_signal`): the @@ -326,6 +343,160 @@ pub fn close_pidfd(fd: i32) { let _ = fd; } +/// Resolve once the process behind `pid` has exited, parked on the kernel's +/// exit notification (Linux pidfd readability, macOS kqueue +/// `EVFILT_PROC`/`NOTE_EXIT`), with no timer. A pid that names no process +/// (already exited and reaped) resolves at once. The kernel handle pins +/// the process instance, so a pid recycled after registration is never +/// mistaken for it. +/// +/// # Errors +/// +/// The OS error when the watch cannot register (no pidfd, descriptor +/// exhaustion, an unsupported platform); the caller owns its fallback. +#[cfg(target_os = "linux")] +pub async fn wait_for_exit(pid: u32) -> std::io::Result<()> { + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + use tokio::io::{unix::AsyncFd, Interest}; + + let fd = match open_pidfd(pid) { + Ok(fd) => fd, + Err(error) if error.raw_os_error() == Some(libc::ESRCH) => return Ok(()), + Err(error) => return Err(error), + }; + // SAFETY: `open_pidfd` returned a fresh descriptor this call owns + // (and closes on drop, on every path). + let fd = AsyncFd::with_interest(unsafe { OwnedFd::from_raw_fd(fd) }, Interest::READABLE)?; + // Every wake is confirmed by a zero-timeout poll (AsyncFd readiness + // can be spurious). + loop { + let mut ready = fd.readable().await?; + if let Ok(result) = ready.try_io(|fd| { + let mut probe = libc::pollfd { + fd: fd.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + // SAFETY: polls one descriptor this call owns, without + // blocking. + match unsafe { libc::poll(&raw mut probe, 1, 0) } { + 1 => Ok(()), + 0 => Err(std::io::ErrorKind::WouldBlock.into()), + _ => Err(std::io::Error::last_os_error()), + } + }) { + return result; + } + } +} + +/// macOS: kqueue `EVFILT_PROC`/`NOTE_EXIT` on a pollable kqueue. +/// +/// # Errors +/// +/// The OS error when the watch cannot register; the caller owns its +/// fallback. +#[cfg(target_os = "macos")] +pub async fn wait_for_exit(pid: u32) -> std::io::Result<()> { + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + use tokio::io::{unix::AsyncFd, Interest}; + + // SAFETY: `kqueue()` takes no arguments. + let kq = unsafe { libc::kqueue() }; + if kq < 0 { + return Err(std::io::Error::last_os_error()); + } + // SAFETY: a fresh descriptor this call owns (and closes on drop, on + // every path). + let kq = unsafe { OwnedFd::from_raw_fd(kq) }; + // The block drops the `kevent` value (its `udata` raw pointer is + // `!Send`) before the first `.await`. + { + let change = libc::kevent { + ident: pid as libc::uintptr_t, + filter: libc::EVFILT_PROC, + flags: libc::EV_ADD, + fflags: libc::NOTE_EXIT, + data: 0, + udata: std::ptr::null_mut(), + }; + // SAFETY: registers one change and reads no events; `change` + // outlives the call. With no event space, an attach failure comes + // back as -1/errno (kevent(2)), not an `EV_ERROR` event. + if unsafe { + libc::kevent( + kq.as_raw_fd(), + &raw const change, + 1, + std::ptr::null_mut(), + 0, + std::ptr::null(), + ) + } < 0 + { + // `ESRCH` means the pid names no attachable process (reaped, + // or already past its exit ref-drain - XNU runs the drain + // before the exit knote fires): the exit already happened. + // Every other errno (EMFILE, ENOMEM, ...) is a watch that + // could not register; the caller falls back. + let error = std::io::Error::last_os_error(); + return match error.raw_os_error() { + Some(libc::ESRCH) => Ok(()), + _ => Err(error), + }; + } + } + let kq = AsyncFd::with_interest(kq, Interest::READABLE)?; + // Only the exit knote is registered, so a drained event is the exit. + loop { + let mut ready = kq.readable().await?; + if let Ok(result) = ready.try_io(|kq| { + let zero = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + let mut event = libc::kevent { + ident: 0, + filter: 0, + flags: 0, + fflags: 0, + data: 0, + udata: std::ptr::null_mut(), + }; + // SAFETY: drains one event with a zero timeout; `event` and + // `zero` outlive the call. + match unsafe { + libc::kevent( + kq.as_raw_fd(), + std::ptr::null(), + 0, + &raw mut event, + 1, + &raw const zero, + ) + } { + 1 => Ok(()), + 0 => Err(std::io::ErrorKind::WouldBlock.into()), + _ => Err(std::io::Error::last_os_error()), + } + }) { + return result; + } + } +} + +/// No kernel exit watch on this platform: the caller takes its own +/// liveness-poll fallback. +/// +/// # Errors +/// +/// Always `Unsupported`. +#[cfg(not(any(target_os = "linux", target_os = "macos")))] +#[allow(clippy::unused_async)] // one awaitable signature across the kernel-watch arms +pub async fn wait_for_exit(_pid: u32) -> std::io::Result<()> { + Err(std::io::ErrorKind::Unsupported.into()) +} + /// Windows: the shared handle probe (win32 has no unreaped-zombie state, /// so existing and running are the same predicate - Node's `kill(pid, 0)` /// checks the same `STILL_ACTIVE` exit code). A query that fails outright @@ -408,3 +579,27 @@ mod windows_tests { assert!(!kill_process_group_or_pid(-1)); } } + +/// The kernel-watch arms' shared exit contract: a pid that names no live +/// process resolves at once, so a worker that died before its watch +/// registered still reports an exit. Both kernel-watch platforms only +/// (the other targets take the caller's fallback by design). +#[cfg(all(test, any(target_os = "linux", target_os = "macos")))] +mod exit_wait_tests { + use super::*; + + #[tokio::test] + async fn wait_for_exit_resolves_at_once_for_a_reaped_pid() { + let mut child = std::process::Command::new("sleep") + .arg("600") + .spawn() + .expect("spawn sleep"); + let pid = child.id(); + child.kill().expect("kill sleep"); + child.wait().expect("reap sleep"); + assert!( + wait_for_exit(pid).await.is_ok(), + "a reaped pid resolves at once" + ); + } +} diff --git a/crates/pa-daemon/src/boot_reap.rs b/crates/pa-daemon/src/boot_reap.rs index 7651262a6..e46390666 100644 --- a/crates/pa-daemon/src/boot_reap.rs +++ b/crates/pa-daemon/src/boot_reap.rs @@ -311,7 +311,7 @@ async fn stop_target_within( // first, then verify the pid still names our process, and only then // does any signal ride the held fd (a signal through this fd can // reach the pinned process and nothing else, ever). - let Some(pidfd) = pa_core::platform::process::open_pidfd(target.pid) else { + let Ok(pidfd) = pa_core::platform::process::open_pidfd(target.pid) else { // The kernel-held handle is unavailable (an unsupported platform, // an old kernel, or a process that just exited): the conservative // default never signals - a missed reap is recoverable, a wrong @@ -945,7 +945,7 @@ mod tests { .expect("spawn sleep"); let pid = child.id(); assert!( - pa_core::platform::process::open_pidfd(pid).is_some(), + pa_core::platform::process::open_pidfd(pid).is_ok(), "the kernel-held handle opens" ); let outcome = stop_target(&target(pid)).await; diff --git a/crates/pa-daemon/src/supervisor/supervision.rs b/crates/pa-daemon/src/supervisor/supervision.rs index 4b2c2e1cf..d3da85e88 100644 --- a/crates/pa-daemon/src/supervisor/supervision.rs +++ b/crates/pa-daemon/src/supervisor/supervision.rs @@ -20,6 +20,9 @@ pub(super) const MAX_CONSECUTIVE_FAILURES: u32 = 5; pub(super) const STABLE_LIFETIME_MS: u64 = 30_000; const BASE_BACKOFF_MS: u64 = 250; const MAX_BACKOFF_MS: u64 = 30_000; +/// The adopted-worker liveness poll used only where the kernel exit watch +/// cannot register (no pidfd, descriptor exhaustion, Windows). +const ADOPTED_EXIT_FALLBACK_POLL: Duration = Duration::from_secs(30); impl Supervisor { /// Watch a worker process: on unexpected exit, restart with backoff. @@ -71,24 +74,30 @@ impl Supervisor { return; } } else if adopted_pid != 0 { - // Adopted worker: poll liveness (cannot wait on a foreign - // pid). A previous relaunch that produced no worker leaves - // pid 0 here - there is nothing to watch, and polling pid 0 - // would report a phantom exit; fall straight to the - // failure/backoff/relaunch arm instead. - loop { - if self.shutting_down.load(Ordering::SeqCst) - || resident.intentional_stop.load(Ordering::SeqCst) - { - return; - } - if !matches!(is_process_alive(adopted_pid as u32), Ok(true)) { - break; + // Adopted worker (no Child handle): park on the kernel's + // exit notification for the pid. A previous relaunch that + // produced no worker leaves pid 0 here - there is nothing + // to watch, and a watch on pid 0 would report a phantom + // exit; fall straight to the failure/backoff/relaunch arm + // instead. + if let Err(error) = + pa_core::platform::process::wait_for_exit(adopted_pid as u32).await + { + // The slow fallback, only where the kernel exit watch + // cannot register. + self.log_line(&format!( + "session worker {}: kernel exit watch unavailable ({error}); polling liveness every {}s", + resident.worker_id, + ADOPTED_EXIT_FALLBACK_POLL.as_secs() + )); + while matches!(is_process_alive(adopted_pid as u32), Ok(true)) { + tokio::time::sleep(ADOPTED_EXIT_FALLBACK_POLL).await; } - tokio::time::sleep(Duration::from_secs(1)).await; } - if resident.intentional_stop.load(Ordering::SeqCst) - || self.shutting_down.load(Ordering::SeqCst) + // A relaunch elsewhere (update abandon, retry) took over the + // resident with a new process. + if self.is_stopping(&resident) + || resident.descriptor.lock().await.pid != adopted_pid { return; } diff --git a/crates/pa-daemon/src/supervisor/tests.rs b/crates/pa-daemon/src/supervisor/tests.rs index 910840d41..854a4db07 100644 --- a/crates/pa-daemon/src/supervisor/tests.rs +++ b/crates/pa-daemon/src/supervisor/tests.rs @@ -169,6 +169,114 @@ async fn a_stop_during_the_storm_leaves_the_terminal_state_to_the_stop() { ); } +/// An adopted worker's watch parks on the kernel's exit notification and +/// hands the exit to whoever owns the resident now - a stop that landed, +/// or a relaunch elsewhere that took the resident's pid. Both cases return +/// at the exit without entering the crash arm, and with no timer in the +/// runtime the paused clock cannot advance. +#[cfg(any(target_os = "linux", target_os = "macos"))] // the kernel-watch arms +#[tokio::test(start_paused = true)] +async fn an_adopted_watch_hands_off_at_exit_without_a_timer() { + let dir = tempfile::TempDir::new().unwrap(); + let agent_dir = dir.path().join("agent"); + std::fs::create_dir_all(&agent_dir).unwrap(); + // A hand-off the watch misses (the unfixed tree) reaches the crash + // arm: its relaunch must fail deterministically (the logs dir cannot + // be created - a file stands where it would go) instead of spawning + // the test binary as a worker. + std::fs::write(agent_dir.join("logs"), "not a directory").unwrap(); + let supervisor = Arc::new( + Supervisor::new(SupervisorOptions { + socket_path: dir.path().join("daemon.sock"), + agent_dir: agent_dir.clone(), + }) + .expect("supervisor"), + ); + { + // A stop that landed owns the exit's terminal state. + let mut worker = std::process::Command::new("sleep") + .arg("600") + .spawn() + .expect("spawn sleep"); + let resident = adopted_watch_resident(dir.path(), worker.id()); + let started = tokio::time::Instant::now(); + let mut watch = std::pin::pin!(Arc::clone(&supervisor).watch_worker( + Arc::clone(&resident), + None, + u64::from(worker.id()) + )); + assert!( + futures::poll!(watch.as_mut()).is_pending(), + "a live adopted worker keeps the watch parked" + ); + resident.intentional_stop.store(true, Ordering::SeqCst); + worker.kill().expect("kill sleep"); + worker.wait().expect("reap sleep"); + watch.await; + assert_eq!( + tokio::time::Instant::now(), + started, + "the kernel exit notification woke the watch; no poll timer fired" + ); + } + { + // A relaunch elsewhere took the resident: the pid no longer names + // the watched process, so the watch hands off instead of running + // the crash arm over the replacement. + let mut worker = std::process::Command::new("sleep") + .arg("600") + .spawn() + .expect("spawn sleep"); + let resident = adopted_watch_resident(dir.path(), worker.id()); + let started = tokio::time::Instant::now(); + let mut watch = std::pin::pin!(Arc::clone(&supervisor).watch_worker( + Arc::clone(&resident), + None, + u64::from(worker.id()) + )); + assert!( + futures::poll!(watch.as_mut()).is_pending(), + "a live adopted worker keeps the watch parked" + ); + resident.descriptor.lock().await.pid = u64::from(std::process::id()); + worker.kill().expect("kill sleep"); + worker.wait().expect("reap sleep"); + watch.await; + assert_eq!( + tokio::time::Instant::now(), + started, + "the kernel exit notification woke the watch; no poll timer fired" + ); + } +} + +/// The watched adopted worker's resident, with the descriptor carrying the +/// watched pid (the post-exit ownership check compares against it). +#[cfg(any(target_os = "linux", target_os = "macos"))] +fn adopted_watch_resident(dir: &std::path::Path, pid: u32) -> Arc { + let descriptor: DaemonWorkerDescriptor = serde_json::from_value(serde_json::json!({ + "version": 2, + "workerId": "w-adopted-watch", + "pid": pid, + "socketPath": "/tmp/none.sock", + "recoveryJournalPath": dir.join("journal.jsonl").to_string_lossy(), + "supervisorSocketPath": "/tmp/none.sock", + "authenticationToken": "token", + "rootActiveSessionId": "w-adopted-watch", + "createdAt": "t", + "updatedAt": "t", + "lifecycle": "ready", + "createCommand": {}, + "consecutiveFailures": 0, + })) + .expect("descriptor"); + ResidentWorker::new( + "w-adopted-watch".to_string(), + descriptor, + dir.join("w-adopted-watch.descriptor.json"), + ) +} + /// The saved-session surfaces (the `list --all` summary row and the /// `list_saved_sessions` catalog row) carry the persisted thinking /// level: the agents-view Model column renders "model:level" for