agentmux_srv/backend/process_tracker/

registry.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

//! `AgentProcessRegistry` — app-wide map from block_id → tracker handle.
//!
//! Created once at host startup, passed into each `SubprocessController`
//! / `PersistentSubprocessController` instance so it can wrap spawns
//! in its per-block job. Polled periodically from a background task
//! that emits `agent:process-added` / `agent:process-exited` events to
//! the frontend's swarm activity panel.
//!
//! Centralizing this here (rather than one tracker per controller)
//! means the lifetime of the tracker matches the lifetime of the pane
//! — multiple turns on the same block share the same job, so
//! descendants from turn N are still visible on turn N+1.

use std::collections::{HashMap, HashSet};
use std::sync::{Arc, OnceLock};

use parking_lot::Mutex;

use super::{agent_started, new_tracker, TrackedProcess, TrackerHandle, TrackingConfidence};
use crate::backend::mps;

/// Host-wide registry, set once at startup. Exposed as a global so
/// `SubprocessController` / `PersistentSubprocessController` can reach
/// it without threading an `Arc` through every constructor + test site.
/// Tests that don't initialize it see `None` and silently skip
/// tracker registration — the job-object spawn path no-ops cleanly.
static GLOBAL: OnceLock<Arc<AgentProcessRegistry>> = OnceLock::new();

pub fn set_global(registry: Arc<AgentProcessRegistry>) {
    let _ = GLOBAL.set(registry);
}

pub fn global() -> Option<Arc<AgentProcessRegistry>> {
    GLOBAL.get().cloned()
}

/// Assign a freshly-spawned child PID to `block_id`'s tracker, creating the
/// tracker if needed. The single call site every `Controller` impl should use
/// immediately after its own spawn — no-ops silently if the registry global
/// isn't initialized (tests), and with a warn log if the platform tracker
/// rejects the PID, since this is opportunistic enrichment, not a liveness signal on
/// its own (see `broker::process::ProcessStatus`'s own doc comment).
pub fn track_spawned(block_id: &str, pid: u32) {
    track(block_id, pid, false);
}

fn track(block_id: &str, pid: u32, agent: bool) {
    let Some(registry) = global() else { return };
    if let Err(e) = registry.assign_kind(block_id, pid, agent) {
        tracing::warn!(
            block_id = %block_id,
            pid = pid,
            err = %e,
            "[process-tracker] assign_process failed"
        );
    }
}

/// Put a process srv spawned on an agent's behalf (`Shell()`, `!cmd`) into
/// that agent's tracker, so it dies with the agent like anything the agent
/// started itself (SPEC_AGENT_TEARDOWN_SINGLE_PATH_2026_10_01.md §6.5). Unlike
/// [`track_spawned`] it never creates a tracker (a block with no agent process
/// has nothing to join) and doesn't mark the PID as a root, so the process
/// still counts as something the agent started. Best-effort, like
/// [`track_spawned`].
pub fn track_adopted(block_id: &str, pid: u32) {
    if block_id.is_empty() {
        return;
    }
    let Some(registry) = global() else { return };
    if let Err(e) = registry.adopt(block_id, pid) {
        tracing::warn!(block_id = %block_id, pid = pid, err = %e, "[process-tracker] adopt failed");
    }
}

/// Make a child about to be spawned for `block_id` join its tracker before it
/// execs, so nothing it forks escapes between spawn and [`track_spawned`]
/// (Linux: the agent's cgroup). `create`: start the block's tracker if it has
/// none (an agent's own CLI); otherwise join only an existing one (`Shell()`,
/// `!cmd`, like [`track_adopted`]). Still call `track_spawned*` /
/// `track_adopted` after the spawn: that records the PID and covers platforms
/// with no pre-exec placement.
pub fn place_spawn(block_id: &str, cmd: &mut tokio::process::Command, create: bool) {
    #[cfg(target_os = "linux")]
    if let Some(target) = spawn_target(block_id, create) {
        super::cgroup_linux::join_before_exec(cmd, target);
    }
    #[cfg(not(target_os = "linux"))]
    let _ = (block_id, cmd, create);
}

/// How a spawn joins its block's tracker (see [`spawn_tracked`]).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Join {
    /// The agent's own process: starts the block's tracker if needed and is
    /// recorded as a root ([`track_spawned_agent`]).
    Agent,
    /// Started on an agent's behalf (`Shell()`, `!cmd`): joins an existing
    /// tracker only ([`track_adopted`]).
    Adopted,
}

/// Spawn `cmd` for `block_id` already inside the block's tracker, so nothing
/// it starts can escape between spawn and assignment: on Linux it joins the
/// cgroup before exec ([`place_spawn`]); on Windows it starts suspended, is
/// assigned to the job, then resumed. The one way an agent-side tokio spawn
/// should start (the Codex App Server, which spawns internally, uses
/// [`place_spawn`] and [`track_spawned_agent`]).
pub fn spawn_tracked(block_id: &str, cmd: &mut tokio::process::Command, join: Join) -> std::io::Result<tokio::process::Child> {
    place_spawn(block_id, cmd, join == Join::Agent);
    #[cfg(windows)]
    let suspended = start_suspended(block_id, cmd, join == Join::Agent);
    let child = cmd.spawn()?;
    if let Some(pid) = child.id() {
        match join {
            Join::Agent => track_spawned_agent(block_id, pid),
            Join::Adopted => track_adopted(block_id, pid),
        }
        // Resumed whether or not the assignment worked: a child left
        // suspended would never run.
        #[cfg(windows)]
        if suspended {
            if let Err(e) = agentmux_common::win32::resume_main_thread(pid) {
                let mut child = child;
                let _ = child.start_kill();
                return Err(std::io::Error::other(format!("resume suspended child {pid}: {e}")));
            }
        }
    }
    Ok(child)
}

/// Make `cmd` start suspended when the block has (or, `create`, gets) a Job
/// Object to put it in first. Agent-side spawns set no creation flag but
/// `CREATE_NO_WINDOW`, which this keeps.
#[cfg(windows)]
fn start_suspended(block_id: &str, cmd: &mut tokio::process::Command, create: bool) -> bool {
    use windows_sys::Win32::System::Threading::{CREATE_NO_WINDOW, CREATE_SUSPENDED};
    let Some(registry) = global() else { return false };
    if block_id.is_empty() {
        return false;
    }
    let tracker = if create {
        Some(registry.ensure_tracker_kind(block_id, true))
    } else {
        registry.inner.lock().get(block_id).map(|e| e.tracker.clone())
    };
    if !tracker.is_some_and(|t| t.confidence() == TrackingConfidence::High) {
        return false;
    }
    cmd.creation_flags(CREATE_NO_WINDOW | CREATE_SUSPENDED);
    true
}

/// [`place_spawn`] for an agent's PTY spawn, which has no pre-exec hook: the
/// command is wrapped to join first (`cgroup_linux::join_before_exec_pty`).
pub fn place_pty_spawn(block_id: &str, cmd: &mut portable_pty::CommandBuilder) {
    #[cfg(target_os = "linux")]
    if let Some(target) = spawn_target(block_id, true) {
        super::cgroup_linux::join_before_exec_pty(cmd, target);
    }
    #[cfg(not(target_os = "linux"))]
    let _ = (block_id, cmd);
}

/// Where a spawn for `block_id` joins its agent's tree, if anywhere.
#[cfg(target_os = "linux")]
fn spawn_target(block_id: &str, create: bool) -> Option<std::path::PathBuf> {
    if block_id.is_empty() {
        return None;
    }
    let registry = global()?;
    let tracker = if create {
        Some(registry.ensure_tracker_kind(block_id, true))
    } else {
        registry.inner.lock().get(block_id).map(|e| e.tracker.clone())
    };
    tracker?.spawn_target()
}

/// Settings key: run agents' process trees at below-normal CPU priority so
/// their builds and tests yield to AgentMux (and the user's foreground apps)
/// under contention, while still using every idle core. Default on.
pub const SETTING_AGENT_BELOW_NORMAL_PRIORITY: &str = "agent:belownormalpriority";

/// [`track_spawned`] for an **agent** process: also applies the agent CPU
/// priority policy to the block's whole tree. Terminal panes a human types in
/// keep calling [`track_spawned`] and are never deprioritized.
pub fn track_spawned_agent(block_id: &str, pid: u32) {
    track(block_id, pid, true);
    let Some(registry) = global() else { return };
    let on = registry.agent_below_normal_priority();
    if let Err(e) = registry.set_below_normal_priority(block_id, on) {
        tracing::warn!(block_id = %block_id, err = %e, "[process-tracker] could not set job CPU priority");
    }
}

/// One block's tracked tree, as [`AgentProcessRegistry::members_by_block`]
/// reports it.
#[derive(Debug, Clone)]
pub struct BlockMembers {
    pub block_id: String,
    pub pids: Vec<u32>,
    /// The PIDs assigned as the block's own process (the agent CLI or shell).
    pub roots: HashSet<u32>,
    pub confidence: TrackingConfidence,
    pub cpu_time_ns: Option<u64>,
}

pub struct AgentProcessRegistry {
    inner: Mutex<HashMap<String, RegistryEntry>>,
    broker: Option<Arc<mps::Broker>>,
    /// Live settings, read per spawn so a settings.json change applies to the
    /// next agent spawn. `None` in tests and before bootstrap wires it.
    config: Mutex<Option<Arc<crate::backend::wconfig::ConfigState>>>,
}

struct RegistryEntry {
    tracker: Arc<dyn TrackerHandle>,
    /// Last-known PID set from the most recent poll. Used to diff
    /// against the current set so we only emit events for
    /// additions/removals — not on every poll tick.
    last_pids: HashSet<u32>,
    /// Every PID handed to `assign` — the agent CLI itself, one per
    /// (re)spawn. `agent_started` excludes them and their non-shell
    /// children (MCP servers) from what we report.
    roots: HashSet<u32>,
}

impl AgentProcessRegistry {
    pub fn new(broker: Option<Arc<mps::Broker>>) -> Self {
        Self {
            inner: Mutex::new(HashMap::new()),
            broker,
            config: Mutex::new(None),
        }
    }

    /// Attach the live settings (bootstrap).
    pub fn set_config(&self, config: Arc<crate::backend::wconfig::ConfigState>) {
        *self.config.lock() = Some(config);
    }

    /// The `agent:belownormalpriority` setting; `true` when unset or when no
    /// settings are attached.
    pub fn agent_below_normal_priority(&self) -> bool {
        self.config
            .lock()
            .as_ref()
            .and_then(|c| c.get_settings().extra.get(SETTING_AGENT_BELOW_NORMAL_PRIORITY).and_then(|v| v.as_bool()))
            .unwrap_or(true)
    }

    /// Apply (or lift) the below-normal priority limit on `block_id`'s tree.
    pub fn set_below_normal_priority(&self, block_id: &str, on: bool) -> Result<(), String> {
        let tracker = self.ensure_tracker(block_id);
        tracker.set_below_normal_priority(on)
    }

    /// Ensure a tracker exists for this block. Idempotent — calling
    /// twice for the same block returns the existing tracker so the
    /// job survives controller re-creation (e.g. on /clear).
    pub fn ensure_tracker(&self, block_id: &str) -> Arc<dyn TrackerHandle> {
        self.ensure_tracker_kind(block_id, false)
    }

    /// [`ensure_tracker`](Self::ensure_tracker) for an agent's block
    /// (`agent`) or a terminal a person types in. The kind matters on every
    /// Unix, where an agent gets a cgroup or the best-effort scan tracker (its
    /// whole tree ends with it) and a terminal keeps Unix semantics: what the
    /// user `nohup`s outlives it.
    pub fn ensure_tracker_kind(&self, block_id: &str, agent: bool) -> Arc<dyn TrackerHandle> {
        let mut map = self.inner.lock();
        if let Some(entry) = map.get(block_id) {
            return entry.tracker.clone();
        }
        let tracker = new_tracker(block_id, agent);
        map.insert(
            block_id.to_string(),
            RegistryEntry {
                tracker: tracker.clone(),
                last_pids: HashSet::new(),
                roots: HashSet::new(),
            },
        );
        tracing::info!(
            block_id = %block_id,
            confidence = ?tracker.confidence(),
            "[process-tracker] registered tracker"
        );
        tracker
    }

    /// Assign a freshly-spawned agent/shell PID to `block_id`'s tracker
    /// (creating it if needed) and remember it as a root.
    /// The tracker is an agent's or a terminal's (see
    /// [`ensure_tracker_kind`](Self::ensure_tracker_kind)).
    pub fn assign_kind(&self, block_id: &str, pid: u32, agent: bool) -> Result<(), String> {
        let tracker = self.ensure_tracker_kind(block_id, agent);
        tracker.assign_process(pid)?;
        if let Some(entry) = self.inner.lock().get_mut(block_id) {
            entry.roots.insert(pid);
        }
        Ok(())
    }

    /// Add `pid` to `block_id`'s tracker if it has one, without marking it a
    /// root. `Ok(false)` when the block has no tracker.
    pub fn adopt(&self, block_id: &str, pid: u32) -> Result<bool, String> {
        let tracker = match self.inner.lock().get(block_id) {
            Some(e) => e.tracker.clone(),
            None => return Ok(false),
        };
        tracker.assign_process(pid)?;
        Ok(true)
    }

    /// Drop a block's tracker — call when the pane closes. The tracker's
    /// Drop impl kills the whole process tree (`KILL_ON_JOB_CLOSE` on
    /// Windows, `cgroup.kill` on Linux; nothing on the stub).
    pub fn remove(&self, block_id: &str) {
        let mut map = self.inner.lock();
        if map.remove(block_id).is_some() {
            tracing::info!(block_id = %block_id, "[process-tracker] dropped tracker on pane close");
        }
    }

    /// `(tracked block count, summed last-known PID count across all
    /// blocks)` — diagnostic-only, for the periodic `mem_attribution` log
    /// (`sysinfo.rs`) to correlate against commit/handle growth. A block
    /// count that grows without bound (no matching `remove` on pane close)
    /// or a PID count far exceeding what's actually open would both be
    /// visible here well before the commit-side symptom is undeniable.
    #[cfg(any(windows, test))]
    pub fn stats(&self) -> (usize, usize) {
        let map = self.inner.lock();
        let pids: usize = map.values().map(|e| e.last_pids.len()).sum();
        (map.len(), pids)
    }

    /// Processes the agent started through its tools (see
    /// [`agent_started`]) — for the RPC endpoint, Swarm and the pane-close
    /// confirmation. Excludes the CLI itself, conhost and MCP servers;
    /// `kill_tree` still takes all of them.
    pub fn list_block(&self, block_id: &str) -> Vec<TrackedProcess> {
        let (tracker, roots) = match self.inner.lock().get(block_id) {
            Some(e) => (e.tracker.clone(), e.roots.clone()),
            None => return Vec::new(),
        };
        agent_started(tracker.list_members(), &roots)
    }

    /// Every tracked block's members for Tower: PIDs (plumbing included), the
    /// roots, the tracker's confidence and the tree's CPU account. The trackers
    /// are queried outside the registry lock.
    pub fn members_by_block(&self) -> Vec<BlockMembers> {
        let entries: Vec<(String, Arc<dyn TrackerHandle>, HashSet<u32>)> = self
            .inner
            .lock()
            .iter()
            .map(|(id, e)| (id.clone(), e.tracker.clone(), e.roots.clone()))
            .collect();
        entries
            .into_iter()
            .map(|(block_id, tracker, roots)| BlockMembers {
                pids: tracker.member_pids(),
                cpu_time_ns: tracker.cpu_time_ns(),
                confidence: tracker.confidence(),
                block_id,
                roots,
            })
            .collect()
    }

    /// Confidence of a block's tracker — drives the "tracking is
    /// best-effort on macOS" badge in the swarm UI.
    pub fn confidence_of(&self, block_id: &str) -> TrackingConfidence {
        self.inner
            .lock()
            .get(block_id)
            .map(|e| e.tracker.confidence())
            .unwrap_or(TrackingConfidence::None)
    }

    /// Ask everything in `block_id`'s tree to exit (SIGTERM on Linux), the
    /// graceful step before [`remove`](Self::remove) kills what's left.
    /// `false` when nothing was asked (no tracker, or no graceful signal).
    pub fn terminate(&self, block_id: &str) -> bool {
        let tracker = self.inner.lock().get(block_id).map(|e| e.tracker.clone());
        tracker.is_some_and(|t| t.terminate())
    }

    /// Every process in `block_id`'s tree, plumbing included; 0 without a
    /// tracker.
    pub fn member_count(&self, block_id: &str) -> usize {
        let tracker = self.inner.lock().get(block_id).map(|e| e.tracker.clone());
        tracker.map_or(0, |t| t.member_count())
    }

    /// Kill the entire process tree for a given block. Returns `true`
    /// if a tracker was found (the kill was dispatched). Does NOT
    /// synchronously wait for descendants to actually exit — the
    /// poller's next tick will pick up the state changes and emit
    /// `agent:process-exited` events the frontend can react to.
    pub fn kill_tree(&self, block_id: &str) -> bool {
        let tracker = self.inner.lock().get(block_id).map(|e| e.tracker.clone());
        match tracker {
            Some(t) => {
                t.kill_tree();
                true
            }
            None => false,
        }
    }

    /// Kill a single PID if it's a member of the given block's tree.
    /// Returns `true` if the tracker was found and the PID matched.
    pub fn kill_pid(&self, block_id: &str, pid: u32) -> bool {
        let tracker = self.inner.lock().get(block_id).map(|e| e.tracker.clone());
        match tracker {
            Some(t) => t.kill_pid(pid),
            None => false,
        }
    }

    /// Poll every tracked block's membership and diff against the
    /// last-known set. Emits `agent:process-added` / `-exited` events
    /// for each delta.
    ///
    /// Called by a background Tokio task on a ~2s interval.
    pub fn poll_and_emit(&self) {
        let mut map = self.inner.lock();
        for (block_id, entry) in map.iter_mut() {
            let current_members = agent_started(entry.tracker.list_members(), &entry.roots);
            let current_pids: HashSet<u32> = current_members.iter().map(|p| p.pid).collect();

            for added_pid in current_pids.difference(&entry.last_pids) {
                if let Some(p) = current_members.iter().find(|p| p.pid == *added_pid) {
                    if let Some(broker) = &self.broker {
                        let added = crate::backend::mps_events::ProcessAddedEvent { block_id: block_id.clone(), process: p.clone() };
                        broker.publish_event::<crate::backend::mps_events::AgentProcessAdded>(&added, vec![format!("block:{}", block_id)], 0);
                    }
                }
            }
            for removed_pid in entry.last_pids.difference(&current_pids) {
                if let Some(broker) = &self.broker {
                    let exited = crate::backend::mps_events::ProcessExitedEvent { block_id: block_id.clone(), pid: *removed_pid };
                    broker.publish_event::<crate::backend::mps_events::AgentProcessExited>(&exited, vec![format!("block:{}", block_id)], 0);
                }
            }

            entry.last_pids = current_pids;
        }
    }

}

/// Spawn the polling task. Drops when the registry's `Arc` refcount
/// hits zero (host shutdown). ~2s cadence balances latency (new
/// processes show up fast) with CPU overhead (job queries are cheap
/// but not free).
pub fn spawn_poller(registry: Arc<AgentProcessRegistry>) {
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(std::time::Duration::from_secs(2));
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            interval.tick().await;
            // Per-process OS queries for every tracked block plus a
            // system-wide process snapshot, under two mutexes — never on an
            // async worker (see `backend::blocking`). Awaited, so still one
            // pass at a time.
            let registry = Arc::clone(&registry);
            if let Err(e) = tokio::task::spawn_blocking(move || registry.poll_and_emit()).await {
                tracing::warn!(error = %e, "[process-tracker] poll task failed");
            }
        }
    });
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn agent_cpu_priority_policy_defaults_on_and_follows_the_setting() {
        let reg = AgentProcessRegistry::new(None);
        // No settings attached (tests, early boot): on.
        assert!(reg.agent_below_normal_priority());

        let config = Arc::new(crate::backend::wconfig::ConfigState::with_config(
            crate::backend::wconfig::build_default_config(),
        ));
        reg.set_config(Arc::clone(&config));
        // Key absent: on.
        assert!(reg.agent_below_normal_priority());

        let mut settings = config.get_settings();
        settings.extra.insert(SETTING_AGENT_BELOW_NORMAL_PRIORITY.to_string(), serde_json::json!(false));
        config.update_settings(settings.clone());
        assert!(!reg.agent_below_normal_priority(), "read live, not captured");

        settings.extra.insert(SETTING_AGENT_BELOW_NORMAL_PRIORITY.to_string(), serde_json::json!(true));
        config.update_settings(settings);
        assert!(reg.agent_below_normal_priority());
    }

    #[test]
    fn track_spawned_is_a_safe_no_op_without_a_global_registry() {
        // `GLOBAL` is only ever set by `bootstrap/network.rs` at real host
        // startup — never in this test binary — so this exercises the exact
        // "tests silently skip tracker registration" behavior the doc comment
        // on `GLOBAL` promises. Must not panic.
        track_spawned("test-block-track-spawned-no-global", 999_999);
    }

    #[test]
    fn stats_reflects_tracked_block_count_and_drops_on_remove() {
        // A fresh, non-global registry (not `global()`) — this counts
        // entries this test itself created, not whatever other tests
        // running in parallel have registered against the process-wide
        // `GLOBAL`. `stats()` backs the `mem_attribution` diagnostic log
        // (sysinfo.rs) added while investigating a 2026-09-20 leak report.
        let registry = AgentProcessRegistry::new(None);
        assert_eq!(registry.stats(), (0, 0), "empty registry");

        registry.ensure_tracker("block-a");
        assert_eq!(registry.stats(), (1, 0), "one block, no poll yet so no known pids");

        registry.ensure_tracker("block-b");
        assert_eq!(registry.stats().0, 2, "two distinct blocks tracked");

        // Re-registering the same block is idempotent (per ensure_tracker's
        // own doc comment) — must not double-count.
        registry.ensure_tracker("block-a");
        assert_eq!(registry.stats().0, 2, "re-registering an existing block does not grow the count");

        registry.remove("block-a");
        assert_eq!(registry.stats().0, 1, "removed block no longer counted");
    }

    // This test used to be `#[ignore]`d as "environment-dependent", blaming
    // the agent-shell harness's own job object. That was wrong: the Windows
    // tracker opened the process with 0x0200 (PROCESS_SET_INFORMATION) where
    // AssignProcessToJobObject needs PROCESS_SET_QUOTA (0x0100), so EVERY
    // assignment failed with Access Denied — in the shipped app too (no
    // production log ever shows "assigned process to job"). Nested jobs are
    // fine; the access mask wasn't. Keep this test running so it can't
    // silently break again. Windows-only: elsewhere `new_tracker` is still
    // the membership-less stub, so there is nothing to observe (Codex P1 on
    // #3425).
    #[test]
    #[cfg(windows)]
    fn ensure_tracker_and_assign_process_track_a_real_short_lived_child() {
        // Uses a fresh, non-global `AgentProcessRegistry` (not `track_spawned`'s
        // `global()` path) so this doesn't touch the process-wide `GLOBAL`
        // OnceLock other tests in this binary may rely on being unset.
        //
        // Spawns a real, disposable child (rather than assigning the test
        // process's own PID) because `JobObjectTracker`'s Drop impl closes the
        // job handle, which fires `KILL_ON_JOB_CLOSE` on Windows — assigning
        // the test runner's own PID would kill the test process the moment
        // the registry (and its tracker) drops at the end of this test.
        // Alive for a few seconds, so it can't exit (and make the assignment
        // race a dead process) before `assign_process` runs; killed below.
        let mut child = if cfg!(windows) {
            std::process::Command::new("cmd")
                .args(["/C", "ping -n 2 127.0.0.1 > nul & ping -n 10 127.0.0.1 > nul"])
                .spawn()
        } else {
            std::process::Command::new("sh")
                .args(["-c", "sleep 10"])
                .spawn()
        }
        .expect("failed to spawn a disposable test child");
        let pid = child.id();

        let registry = AgentProcessRegistry::new(None);
        registry
            .assign_kind("test-block-real-spawn", pid, false)
            .expect("assign should succeed for a live child we just spawned");

        let raw = registry.ensure_tracker("test-block-real-spawn").list_members();
        assert!(
            raw.iter().any(|p| p.pid == pid),
            "expected pid {pid} to be a job member after assign"
        );

        // The root itself is plumbing, not agent-started; the `ping` its shell
        // launches is. The second ping starts ~1s in, well after assignment,
        // so it can't slip through the spawn/assign race.
        let is_ping = |p: &TrackedProcess| p.command.to_ascii_lowercase().ends_with("ping.exe");
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
        let started = loop {
            let listed = registry.list_block("test-block-real-spawn");
            if listed.iter().any(is_ping) || std::time::Instant::now() > deadline {
                break listed;
            }
            std::thread::sleep(std::time::Duration::from_millis(100));
        };
        assert!(
            !started.iter().any(|p| p.pid == pid),
            "the assigned root must not be listed as agent-started"
        );
        assert!(
            started.iter().any(|p| is_ping(p) && p.parent_pid == Some(pid)),
            "expected the shell's ping child in list_block, got {started:?}"
        );

        // Reap before `registry` drops, so KILL_ON_JOB_CLOSE has nothing left
        // to terminate.
        let _ = child.kill();
        let _ = child.wait();
    }

    #[test]
    fn adopt_never_creates_a_tracker() {
        let registry = AgentProcessRegistry::new(None);
        assert_eq!(registry.adopt("no-such-block", std::process::id()), Ok(false));
        assert_eq!(registry.stats().0, 0, "adopt must not register a tracker");
    }

    /// A `Shell()` process adopted into an agent's tracker is listed as
    /// agent-started and dies when the agent's tracker is released
    /// (agent_teardown spec §6.5).
    #[test]
    #[cfg(windows)]
    fn an_adopted_process_is_listed_and_dies_with_the_agent() {
        let spawn = || {
            std::process::Command::new("cmd")
                .args(["/C", "ping -n 30 127.0.0.1 > nul"])
                .spawn()
                .expect("failed to spawn a disposable test child")
        };
        let (mut agent, mut shell) = (spawn(), spawn());
        let registry = AgentProcessRegistry::new(None);
        registry.assign_kind("test-block-adopt", agent.id(), false).expect("assign the agent root");
        assert_eq!(registry.adopt("test-block-adopt", shell.id()), Ok(true));

        let listed = registry.list_block("test-block-adopt");
        assert!(
            listed.iter().any(|p| p.pid == shell.id()),
            "an adopted shell is agent-started, not a root: {listed:?}"
        );

        registry.remove("test-block-adopt");
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while shell.try_wait().ok().flatten().is_none() && std::time::Instant::now() < deadline {
            std::thread::sleep(std::time::Duration::from_millis(50));
        }
        assert!(shell.try_wait().ok().flatten().is_some(), "releasing the agent's tracker must end the adopted shell");
        let _ = agent.kill();
        let _ = agent.wait();
    }

    /// `spawn_tracked`'s Windows path: a child started suspended is in the
    /// job before it runs, and runs once resumed (it never stays stuck).
    #[tokio::test]
    #[cfg(windows)]
    async fn a_suspended_child_is_assigned_then_resumed_and_runs() {
        use windows_sys::Win32::System::Threading::{CREATE_NO_WINDOW, CREATE_SUSPENDED};
        let tracker = super::super::windows::JobObjectTracker::new("test-suspended").unwrap();
        let mut cmd = tokio::process::Command::new("cmd");
        cmd.args(["/C", "exit 7"]).creation_flags(CREATE_NO_WINDOW | CREATE_SUSPENDED);
        let mut child = cmd.spawn().expect("spawn suspended");
        let pid = child.id().unwrap();
        tracker.assign_process(pid).unwrap();
        assert!(tracker.list_members().iter().any(|p| p.pid == pid), "in the job before it runs");
        agentmux_common::win32::resume_main_thread(pid).unwrap();
        let status = tokio::time::timeout(std::time::Duration::from_secs(10), child.wait())
            .await
            .expect("the resumed child ran and exited")
            .unwrap();
        assert_eq!(status.code(), Some(7));
    }

    /// The job's CPU account grows while a member works and keeps that time
    /// after the member exits, which a per-process sum would lose.
    #[test]
    #[cfg(windows)]
    fn a_jobs_cpu_time_keeps_what_exited_members_used() {
        let tracker = super::super::windows::JobObjectTracker::new("test-job-cpu").unwrap();
        assert_eq!(tracker.cpu_time_ns(), Some(0), "an empty job has used nothing");
        // A busy child: cmd counting to two million (a second or more of CPU),
        // then exiting.
        let mut child = std::process::Command::new("cmd")
            .args(["/C", "for /L %i in (1,1,2000000) do @rem"])
            .spawn()
            .expect("spawn cmd");
        tracker.assign_process(child.id()).unwrap();
        let _ = child.wait();
        let used = tracker.cpu_time_ns().unwrap();
        assert!(used >= 300_000_000, "the exited member's CPU time is kept: {used} ns");
        assert!(tracker.member_pids().is_empty(), "nothing is left running");
    }

    /// After `kill_tree` the job stays usable: a respawned process is still
    /// assigned and listed.
    #[test]
    #[cfg(windows)]
    fn a_job_still_tracks_after_kill_tree() {
        let spawn = || {
            std::process::Command::new("cmd")
                .args(["/C", "ping -n 30 127.0.0.1 > nul"])
                .spawn()
                .expect("failed to spawn a disposable test child")
        };
        let tracker = super::super::windows::JobObjectTracker::new("test-kill-keeps-job").unwrap();
        let mut first = spawn();
        tracker.assign_process(first.id()).unwrap();
        tracker.kill_tree();
        let _ = first.wait();
        let mut second = spawn();
        tracker.assign_process(second.id()).expect("the job still accepts processes");
        assert!(tracker.list_members().iter().any(|p| p.pid == second.id()));
        tracker.kill_tree();
        let _ = second.wait();
    }
}