//! Ownership fence for a subprocess and every member of its process group.
//!
//! The command's immediate child is not a sufficient lifetime fence. A shell
//! can start `server >/dev/null 2>&1 &` and exit immediately: its pipes close
//! and its PID becomes reapable while the server remains in the same process
//! group. Treating that leader status as terminal both lies to `TaskRegistry`
//! and leaves the server outside coordinated shutdown.
//!
//! On Unix, [`ProcessGroupChild::spawn`] therefore starts an independent
//! watchdog as the process-group leader and joins the requested command to its
//! already-live group. The watchdog script and its anchored-group signaling
//! helpers live in `harness::watchdog` — the ONE shared copy for local tasks
//! and terminal sessions, so crash-recovery semantics cannot drift between
//! the two. The watchdog:
//!
//! - pins the PGID until Studio has proved that no other group member exists;
//! - stays alive across an immediate command-leader exit or closed stdio;
//! - receives a parent-liveness pipe, so an uncatchable Studio crash makes it
//!   TERM, then KILL, the remaining group;
//! - ignores TERM itself, allowing graceful TERM to reach the workload while
//!   preserving the ownership anchor for a later KILL/reap.
//!
//! A successful [`ProcessGroupChild::wait`] means the direct child was reaped,
//! no same-PG descendant remained, and the watchdog itself was reaped under the
//! same ownership fence used by signaling. Only that result authorizes a task's
//! terminal transition.

use std::path::Path;
use std::process::{ExitStatus, Output, Stdio};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use tokio::io::AsyncReadExt;
use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};

use crate::tasks::KillSignal;

type SignalGroup = fn(u32, KillSignal) -> bool;

const POLL_INTERVAL: Duration = Duration::from_millis(20);

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum OwnershipState {
    /// The independent group anchor has not been successfully reaped. Its PID
    /// therefore still pins the numeric PGID, even if the command leader has
    /// already exited.
    Armed(Option<u32>),
    /// Successful group-quiescence proof is a one-way transition. Never signal
    /// this number again, even if a later process receives it.
    Reaped(ExitStatus),
    /// The owner was dropped without a completed proof. Its best-effort KILL
    /// ran and every escaped control is permanently inert.
    Released,
}

struct ProcessGroupOwnership {
    state: Mutex<OwnershipState>,
    signal_group: SignalGroup,
}

impl ProcessGroupOwnership {
    fn new(pid: Option<u32>, signal_group: SignalGroup) -> Self {
        Self {
            state: Mutex::new(OwnershipState::Armed(pid)),
            signal_group,
        }
    }

    fn lock(&self) -> MutexGuard<'_, OwnershipState> {
        self.state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// Runs the signal command while holding the ownership fence. A concurrent
    /// successful wait cannot reap/release the anchor until this signal has
    /// completed; conversely, once wait disarms first this fails closed.
    fn signal(&self, signal: KillSignal) -> bool {
        let state = self.lock();
        let OwnershipState::Armed(Some(pid)) = *state else {
            return false;
        };
        (self.signal_group)(pid, signal)
    }

    fn cleanup_on_owner_drop(&self) {
        let mut state = self.lock();
        if let OwnershipState::Armed(pid) = *state {
            if let Some(pid) = pid {
                let _ = (self.signal_group)(pid, KillSignal::Kill);
            }
            *state = OwnershipState::Released;
        }
    }

    #[cfg(test)]
    fn state(&self) -> OwnershipState {
        *self.lock()
    }
}

#[cfg(unix)]
struct GroupAnchor {
    child: Child,
    parent_liveness: Option<ChildStdin>,
    shutdown_started: bool,
}

/// Owns one spawned command leader, its independent group anchor, and all
/// authority to signal their PGID.
pub(crate) struct ProcessGroupChild {
    // Drop runs our explicit `Drop` implementation before fields are dropped,
    // so group cleanup happens while both child handles still pin identities.
    child: Child,
    ownership: Arc<ProcessGroupOwnership>,
    leader_status: Option<ExitStatus>,
    #[cfg(unix)]
    anchor: Option<GroupAnchor>,
}

/// Signal capability for select loops whose wait future already mutably
/// borrows the owning [`ProcessGroupChild`]. It cannot extend signal authority
/// past a successful whole-group reap: both handles share the same one-way
/// ownership state.
pub(crate) struct ProcessGroupControl {
    ownership: Arc<ProcessGroupOwnership>,
}

impl ProcessGroupControl {
    pub(crate) async fn signal(&self, signal: KillSignal) -> bool {
        let ownership = Arc::clone(&self.ownership);
        tokio::task::spawn_blocking(move || ownership.signal(signal))
            .await
            .unwrap_or(false)
    }
}

impl ProcessGroupChild {
    /// Spawns `command` into an independently anchored process group.
    ///
    /// Callers should configure stdio, cwd, environment, and `kill_on_drop`,
    /// but must use this method instead of `Command::spawn`: wrapping an
    /// already-spawned group leader cannot protect the leader-exited descendant
    /// case because that leader itself is the identity that must stay pinned.
    pub(crate) async fn spawn(
        command: &mut Command,
        child_lifetime_lock_path: &Path,
    ) -> std::io::Result<Self> {
        #[cfg(unix)]
        {
            if let Some(parent) = child_lifetime_lock_path
                .parent()
                .filter(|parent| !parent.as_os_str().is_empty())
            {
                // Deep callers (notably Git helpers and freshly materialized
                // sandbox registries) may select the app-wide fence before
                // anything else has created its directory. Do this on Tokio's
                // filesystem pool before the synchronous anchor setup so a
                // valid spawn cannot fail merely because its retention path
                // has not been touched yet.
                tokio::fs::create_dir_all(parent).await?;
            }
            let (mut anchor_child, parent_liveness, group_id) =
                spawn_group_anchor(child_lifetime_lock_path)?;
            let process_group = i32::try_from(group_id)
                .map_err(|_| std::io::Error::other("watchdog pid does not fit a process group"))?;
            command.process_group(process_group);
            let child = match command.spawn() {
                Ok(child) => child,
                Err(error) => {
                    drop(parent_liveness);
                    // No workload was spawned. Let the watchdog prove that
                    // its group is empty before it exits and releases the
                    // inherited shared child-lifetime fence.
                    let _ = anchor_child.wait().await;
                    return Err(error);
                }
            };
            let ownership = Arc::new(ProcessGroupOwnership::new(
                Some(group_id),
                signal_anchored_process_group,
            ));
            Ok(Self {
                child,
                ownership,
                leader_status: None,
                anchor: Some(GroupAnchor {
                    child: anchor_child,
                    parent_liveness: Some(parent_liveness),
                    shutdown_started: false,
                }),
            })
        }

        #[cfg(not(unix))]
        {
            let child = command.spawn()?;
            Ok(Self::new(child))
        }
    }

    /// Unanchored constructor retained for non-Unix and ownership-fence unit
    /// tests. Production Unix route/setup spawns must use [`Self::spawn`].
    #[cfg(not(unix))]
    fn new(child: Child) -> Self {
        Self::new_with_signaller(child, signal_process_group)
    }

    #[cfg(any(test, not(unix)))]
    fn new_with_signaller(child: Child, signal_group: SignalGroup) -> Self {
        let ownership = Arc::new(ProcessGroupOwnership::new(child.id(), signal_group));
        Self {
            child,
            ownership,
            leader_status: None,
            #[cfg(unix)]
            anchor: None,
        }
    }

    /// Returns the owned process-group id. On Unix this is the independent
    /// anchor's pid, not necessarily the immediate command child's pid.
    pub(crate) fn id(&self) -> Option<u32> {
        match *self.ownership.lock() {
            OwnershipState::Armed(pid) => pid,
            OwnershipState::Reaped(_) | OwnershipState::Released => None,
        }
    }

    pub(crate) fn control(&self) -> ProcessGroupControl {
        ProcessGroupControl {
            ownership: Arc::clone(&self.ownership),
        }
    }

    pub(crate) fn take_stdout(&mut self) -> Option<ChildStdout> {
        self.child.stdout.take()
    }

    pub(crate) fn take_stdin(&mut self) -> Option<ChildStdin> {
        self.child.stdin.take()
    }

    pub(crate) fn take_stderr(&mut self) -> Option<ChildStderr> {
        self.child.stderr.take()
    }

    /// Signals the owned process group if and only if its anchor has not been
    /// successfully reaped. The blocking `kill`/`taskkill` invocation runs off
    /// the async worker thread and is serialized with the final disarm.
    pub(crate) async fn signal(&self, signal: KillSignal) -> bool {
        self.control().signal(signal).await
    }

    /// Waits until the direct leader AND every same-process-group descendant
    /// are gone, then reaps the independent anchor and atomically retires all
    /// signal authority.
    ///
    /// Group-enumeration failures retry forever rather than returning an
    /// unproven terminal result. Keeping this future pending deliberately
    /// leaves the corresponding TaskRegistry entry owned and Running, so a
    /// later Stop/shutdown can retry signaling it.
    pub(crate) async fn wait(&mut self) -> std::io::Result<ExitStatus> {
        loop {
            {
                let state = self.ownership.lock();
                if let OwnershipState::Reaped(status) = *state {
                    return Ok(status);
                }
                if matches!(*state, OwnershipState::Released) {
                    return Err(std::io::Error::other(
                        "process-group owner was already released",
                    ));
                }
            }

            if self.leader_status.is_none() {
                match self.child.try_wait() {
                    Ok(Some(status)) => self.leader_status = Some(status),
                    Ok(None) => {}
                    Err(error) => {
                        tracing::error!(%error, "cannot reap process-group command leader; retaining ownership");
                    }
                }
            }

            #[cfg(unix)]
            let anchored_group = match (self.leader_status, self.anchor.as_ref()) {
                (Some(leader_status), Some(_)) => {
                    let group_id = self.id().ok_or_else(|| {
                        std::io::Error::other("process-group ownership disappeared before reap")
                    })?;
                    Some((leader_status, group_id))
                }
                _ => None,
            };
            #[cfg(unix)]
            if let Some((leader_status, group_id)) = anchored_group {
                match group_has_non_anchor_members(group_id).await {
                    Ok(true) => {}
                    Ok(false) => {
                        // Once only the anchor remains, no process capable of
                        // forking another same-group member exists. Kill/reap
                        // the anchor under the signal fence; that exact
                        // critical section closes the PGID-reuse window.
                        let mut state = self.ownership.lock();
                        if let OwnershipState::Reaped(status) = *state {
                            return Ok(status);
                        }
                        if matches!(*state, OwnershipState::Released) {
                            return Err(std::io::Error::other(
                                "process-group owner was released during reap",
                            ));
                        }
                        let Some(anchor) = self.anchor.as_mut() else {
                            return Err(std::io::Error::other(
                                "process-group anchor disappeared during reap",
                            ));
                        };
                        if !anchor.shutdown_started {
                            let _ = anchor.child.start_kill();
                            anchor.shutdown_started = true;
                        }
                        match anchor.child.try_wait() {
                            Ok(Some(_)) => {
                                drop(anchor.parent_liveness.take());
                                *state = OwnershipState::Reaped(leader_status);
                                return Ok(leader_status);
                            }
                            Ok(None) => {}
                            Err(error) => {
                                tracing::error!(%error, "cannot reap process-group anchor; retaining ownership");
                            }
                        }
                    }
                    Err(error) => {
                        tracing::error!(
                            %error,
                            group_id,
                            "cannot prove process-group quiescence; retaining ownership"
                        );
                    }
                }
            }

            #[cfg(not(unix))]
            if let Some(status) = self.leader_status {
                let mut state = self.ownership.lock();
                *state = OwnershipState::Reaped(status);
                return Ok(status);
            }

            // Unit-only unanchored Unix owners retain the previous
            // direct-child fence semantics. All production Unix call sites use
            // `spawn`, which always installs `anchor`.
            #[cfg(unix)]
            if self.anchor.is_none() {
                if let Some(status) = self.leader_status {
                    let mut state = self.ownership.lock();
                    *state = OwnershipState::Reaped(status);
                    return Ok(status);
                }
            }

            tokio::time::sleep(POLL_INTERVAL).await;
        }
    }

    /// KILLs the group and keeps owning it until [`Self::wait`] proves the
    /// complete reap. `warning_after` is diagnostic only: expiration never
    /// turns an unproven cleanup into a terminal task.
    pub(crate) async fn kill_and_reap(
        &mut self,
        warning_after: Duration,
        context: &'static str,
    ) -> ExitStatus {
        let _ = self.signal(KillSignal::Kill).await;
        match tokio::time::timeout(warning_after, self.wait()).await {
            Ok(Ok(status)) => return status,
            Ok(Err(error)) => {
                tracing::error!(%error, context, "process-group reap failed; retaining ownership");
            }
            Err(_) => {
                tracing::error!(
                    context,
                    "process-group reap exceeded warning deadline; retaining ownership"
                );
            }
        }
        loop {
            match self.wait().await {
                Ok(status) => return status,
                Err(error) => {
                    tracing::error!(%error, context, "process-group reap retry failed; retaining ownership");
                    tokio::time::sleep(POLL_INTERVAL).await;
                }
            }
        }
    }

    /// Borrowing `wait_with_output`: drain both pipes concurrently while
    /// retaining the group owner, then wait for whole-group quiescence.
    pub(crate) async fn wait_with_output(&mut self) -> std::io::Result<Output> {
        let stdout = self.take_stdout();
        let stderr = self.take_stderr();
        let read_stdout = async move {
            let mut bytes = Vec::new();
            if let Some(mut pipe) = stdout {
                pipe.read_to_end(&mut bytes).await?;
            }
            Ok::<_, std::io::Error>(bytes)
        };
        let read_stderr = async move {
            let mut bytes = Vec::new();
            if let Some(mut pipe) = stderr {
                pipe.read_to_end(&mut bytes).await?;
            }
            Ok::<_, std::io::Error>(bytes)
        };
        let (stdout, stderr) = tokio::try_join!(read_stdout, read_stderr)?;
        let status = self.wait().await?;
        Ok(Output {
            status,
            stdout,
            stderr,
        })
    }
}

impl Drop for ProcessGroupChild {
    fn drop(&mut self) {
        // Best effort and intentionally synchronous: Drop may run during
        // runtime teardown. The parent-liveness writer is dropped immediately
        // afterward, waking the independent watchdog if the direct KILL did
        // not complete before the runtime disappeared.
        self.ownership.cleanup_on_owner_drop();
    }
}

/// Opens+locks the shared child-lifetime fence and spawns the parent-liveness
/// watchdog (`harness::watchdog` — the one copy shared with terminal-session)
/// under this crate's own `ps`-visible argv0 label.
#[cfg(unix)]
fn spawn_group_anchor(
    child_lifetime_lock_path: &Path,
) -> std::io::Result<(Child, ChildStdin, u32)> {
    let lifetime_lock = harness::watchdog::open_shared_lifetime_lock(child_lifetime_lock_path)?;
    let anchor =
        harness::watchdog::spawn_anchor("decocms-local-api-watchdog", Some(lifetime_lock))?;
    Ok((anchor.child, anchor.parent_liveness, anchor.group_id))
}

#[cfg(unix)]
async fn group_has_non_anchor_members(group_id: u32) -> std::io::Result<bool> {
    let output = Command::new("pgrep")
        .args(["-g", &group_id.to_string(), "."])
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .kill_on_drop(true)
        .output()
        .await?;
    // `pgrep` exits 1 when it matched nothing. That is a valid empty group
    // observation (e.g. the anchor was already KILLed but remains our unreaped
    // child handle); all other nonzero statuses are indeterminate.
    if !output.status.success() {
        if output.status.code() == Some(1) {
            return Ok(false);
        }
        return Err(std::io::Error::other(format!(
            "pgrep exited {:?}",
            output.status.code()
        )));
    }
    let anchor_pid = group_id.to_string();
    Ok(String::from_utf8_lossy(&output.stdout)
        .lines()
        .map(str::trim)
        .any(|pid| !pid.is_empty() && pid != anchor_pid))
}

/// Signal every current member except the anchor — the shared
/// enumerate-then-signal helper in `harness::watchdog` (see its doc for why
/// group-wide `kill -SIG -<pgid>` is forbidden while the anchor owns the
/// restart fence). The anchor is this group's leader, so its pid IS the
/// group id. Kept as a named `fn` so it fits the [`SignalGroup`] pointer.
#[cfg(unix)]
fn signal_anchored_process_group(group_id: u32, signal: KillSignal) -> bool {
    let signal = match signal {
        KillSignal::Term => harness::watchdog::Signal::Term,
        KillSignal::Kill => harness::watchdog::Signal::Kill,
    };
    harness::watchdog::signal_non_anchor_members(group_id, group_id, signal)
}

#[cfg(not(unix))]
fn signal_process_group(pid: u32, _signal: KillSignal) -> bool {
    std::process::Command::new("taskkill")
        .args(["/PID", &pid.to_string(), "/T", "/F"])
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .is_ok_and(|status| status.success())
}

#[cfg(test)]
mod tests {
    use std::fs::OpenOptions;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use super::*;

    static REUSE_SIGNALS: AtomicUsize = AtomicUsize::new(0);
    static WAIT_SIGNALS: AtomicUsize = AtomicUsize::new(0);
    static DROP_SIGNALS: AtomicUsize = AtomicUsize::new(0);
    static ESCAPED_SIGNALS: AtomicUsize = AtomicUsize::new(0);
    static IGNORED_SIGNALS: AtomicUsize = AtomicUsize::new(0);

    fn record_reuse_signal(_pid: u32, _signal: KillSignal) -> bool {
        REUSE_SIGNALS.fetch_add(1, Ordering::SeqCst);
        true
    }

    fn record_wait_signal(_pid: u32, _signal: KillSignal) -> bool {
        WAIT_SIGNALS.fetch_add(1, Ordering::SeqCst);
        true
    }

    fn record_drop_signal(_pid: u32, _signal: KillSignal) -> bool {
        DROP_SIGNALS.fetch_add(1, Ordering::SeqCst);
        true
    }

    fn record_escaped_signal(_pid: u32, _signal: KillSignal) -> bool {
        ESCAPED_SIGNALS.fetch_add(1, Ordering::SeqCst);
        true
    }

    fn ignore_signal(_pid: u32, _signal: KillSignal) -> bool {
        IGNORED_SIGNALS.fetch_add(1, Ordering::SeqCst);
        true
    }

    #[test]
    fn missing_process_identity_fails_closed() {
        let ownership = ProcessGroupOwnership::new(None, record_reuse_signal);
        assert_eq!(ownership.state(), OwnershipState::Armed(None));
        assert!(!ownership.signal(KillSignal::Term));
    }

    #[test]
    fn reaped_pid_is_never_signaled_even_if_the_number_is_reused() {
        REUSE_SIGNALS.store(0, Ordering::SeqCst);
        let ownership = ProcessGroupOwnership::new(Some(41_337), record_reuse_signal);
        *ownership.lock() = OwnershipState::Reaped(success_status());
        assert!(!ownership.signal(KillSignal::Kill));
        assert_eq!(REUSE_SIGNALS.load(Ordering::SeqCst), 0);
    }

    #[cfg(unix)]
    fn success_status() -> ExitStatus {
        use std::os::unix::process::ExitStatusExt;
        ExitStatus::from_raw(0)
    }

    #[cfg(windows)]
    fn success_status() -> ExitStatus {
        use std::os::windows::process::ExitStatusExt;
        ExitStatus::from_raw(0)
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn spawn_materializes_a_missing_child_lifetime_lock_parent() {
        let dir = tempfile::tempdir().unwrap();
        let lifetime_lock_path = dir
            .path()
            .join("not-yet-created")
            .join("nested")
            .join("child-lifetime.lock");
        let mut command = Command::new("sh");
        command
            .args(["-c", "exit 0"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true);

        let mut owned = ProcessGroupChild::spawn(&mut command, &lifetime_lock_path)
            .await
            .expect("spawn creates the child-lifetime fence parent");
        tokio::time::timeout(Duration::from_secs(2), owned.wait())
            .await
            .expect("fixture group completed")
            .expect("fixture wait succeeded");

        assert!(lifetime_lock_path.is_file());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn leader_exit_and_closed_stdio_do_not_hide_a_live_group_descendant() {
        let dir = tempfile::tempdir().unwrap();
        let pid_file = dir.path().join("descendant.pid");
        let lifetime_lock_path = dir.path().join("child-lifetime.lock");
        let mut command = Command::new("sh");
        command
            .args([
                "-c",
                &format!(
                    "sleep 30 >/dev/null 2>&1 & echo $! > {}",
                    pid_file.display()
                ),
            ])
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        let mut owned = ProcessGroupChild::spawn(&mut command, &lifetime_lock_path)
            .await
            .expect("spawn anchored fixture");

        let descendant = tokio::time::timeout(Duration::from_secs(2), async {
            loop {
                if let Ok(raw) = tokio::fs::read_to_string(&pid_file).await {
                    if let Ok(pid) = raw.trim().parse::<u32>() {
                        break pid;
                    }
                }
                tokio::time::sleep(POLL_INTERVAL).await;
            }
        })
        .await
        .expect("fixture wrote descendant pid");

        assert!(
            tokio::time::timeout(Duration::from_millis(150), owned.wait())
                .await
                .is_err(),
            "leader exit and stdio EOF must not look like whole-group terminal"
        );
        let contender = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&lifetime_lock_path)
            .expect("open lifetime fence contender");
        assert!(
            matches!(contender.try_lock(), Err(std::fs::TryLockError::WouldBlock)),
            "anchor must retain its shared restart fence while a descendant lives"
        );
        assert!(owned.signal(KillSignal::Term).await);
        let status = tokio::time::timeout(Duration::from_secs(3), owned.wait())
            .await
            .expect("TERM reaped background descendant")
            .expect("group reap succeeded");
        assert!(status.success(), "direct shell leader exited successfully");
        assert!(!process_alive(descendant));
        assert!(matches!(owned.ownership.state(), OwnershipState::Reaped(_)));
        let successor = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&lifetime_lock_path)
            .expect("open post-reap lifetime fence");
        successor
            .try_lock()
            .expect("exclusive recovery fence becomes available only after reap");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn cleanup_warning_deadline_does_not_release_an_unreaped_owner() {
        IGNORED_SIGNALS.store(0, Ordering::SeqCst);
        let mut command = Command::new("sh");
        command
            .args(["-c", "sleep 30"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let child = command.spawn().expect("spawn cleanup-timeout fixture");
        let pid = child.id().expect("fixture pid");
        let owned = ProcessGroupChild::new_with_signaller(child, ignore_signal);
        let control = owned.control();
        let cleanup = tokio::spawn(async move {
            let mut owned = owned;
            owned
                .kill_and_reap(Duration::from_millis(25), "test cleanup")
                .await
        });

        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(
            !cleanup.is_finished(),
            "warning timeout must retain the live owner instead of returning terminal"
        );
        assert!(control.signal(KillSignal::Term).await);
        assert!(IGNORED_SIGNALS.load(Ordering::SeqCst) >= 2);

        let _ = std::process::Command::new("kill")
            .args(["-KILL", &format!("-{pid}")])
            .status();
        tokio::time::timeout(Duration::from_secs(2), cleanup)
            .await
            .expect("cleanup completed after a proven reap")
            .expect("cleanup task did not panic");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn panic_after_successful_wait_cannot_run_drop_cleanup() {
        WAIT_SIGNALS.store(0, Ordering::SeqCst);
        let mut command = Command::new("sh");
        command
            .args(["-c", "exit 0"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let child = command.spawn().expect("spawn process-group fixture");
        let mut owned = ProcessGroupChild::new_with_signaller(child, record_wait_signal);
        owned.wait().await.expect("wait fixture");

        let panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _owned = owned;
            panic!("panic after wait");
        }));
        assert!(panic.is_err());
        assert_eq!(WAIT_SIGNALS.load(Ordering::SeqCst), 0);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn dropping_unreaped_owner_runs_group_cleanup_once() {
        DROP_SIGNALS.store(0, Ordering::SeqCst);
        let mut command = Command::new("sh");
        command
            .args(["-c", "sleep 30"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let child = command.spawn().expect("spawn process-group fixture");
        let owned = ProcessGroupChild::new_with_signaller(child, record_drop_signal);
        drop(owned);
        assert_eq!(DROP_SIGNALS.load(Ordering::SeqCst), 1);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn control_that_outlives_owner_is_permanently_inert() {
        ESCAPED_SIGNALS.store(0, Ordering::SeqCst);
        let mut command = Command::new("sh");
        command
            .args(["-c", "sleep 30"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let child = command.spawn().expect("spawn process-group fixture");
        let owned = ProcessGroupChild::new_with_signaller(child, record_escaped_signal);
        let escaped = owned.control();

        drop(owned);
        assert_eq!(ESCAPED_SIGNALS.load(Ordering::SeqCst), 1);
        assert!(!escaped.signal(KillSignal::Term).await);
        assert_eq!(ESCAPED_SIGNALS.load(Ordering::SeqCst), 1);
    }

    #[cfg(unix)]
    fn process_alive(pid: u32) -> bool {
        std::process::Command::new("kill")
            .args(["-0", &pid.to_string()])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status()
            .is_ok_and(|status| status.success())
    }
}
