fix(tests): make cargo test --lib green — PATH leak and projector deadline flakes

src/env.rs

@@ -8,7 +8,7 @@
 use std::ffi::{OsStr, OsString};
 
 #[cfg(any(test, feature = "test-support"))]
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 #[cfg(any(test, feature = "test-support"))]
 use std::sync::{OnceLock, RwLock};
 
@@ -79,14 +79,90 @@ pub fn remove_test_var<K: AsRef<OsStr>>(key: K) {
         .insert(key.as_ref().to_os_string(), None);
 }
 
-/// Resolve an executable only when tests explicitly override `PATH`.
+// Program names that must resolve as missing, regardless of what is really on
+// `PATH`. See [`mask_test_programs`].
+#[cfg(any(test, feature = "test-support"))]
+static MASKED_PROGRAMS: OnceLock<RwLock<HashSet<OsString>>> = OnceLock::new();
+
+/// Directory component that cannot exist, so a masked program spawns with the
+/// same `ErrorKind::NotFound` a genuinely absent binary would produce. Any path
+/// with more than one component bypasses the platform's `PATH` search entirely.
+#[cfg(any(test, feature = "test-support"))]
+const MASKED_PROGRAM_DIR: &str = "cortex-test-masked-program-does-not-exist";
+
+/// Make specific bare program names resolve as missing until the returned guard
+/// is dropped.
+///
+/// This exists so a test asserting "this command is not installed" does not have
+/// to replace `PATH` to say so. The override map is process-global, so replacing
+/// `PATH` with a fixture directory takes every *other* test's subprocess spawns
+/// down with it — they stop finding `sh`, `git`, and everything else — and under
+/// plain `cargo test` those tests run concurrently. Masking states the actual
+/// intent instead, and scopes the blast radius to the named programs.
+///
+/// The mask is still process-global: only mask programs whose spawning tests are
+/// serialized against this one.
+#[cfg(any(test, feature = "test-support"))]
+#[doc(hidden)]
+#[must_use = "the mask is lifted when the guard is dropped"]
+pub fn mask_test_programs<I, S>(names: I) -> MaskedPrograms
+where
+    I: IntoIterator<Item = S>,
+    S: AsRef<OsStr>,
+{
+    let names: Vec<OsString> = names
+        .into_iter()
+        .map(|name| name.as_ref().to_os_string())
+        .collect();
+    let mut masked = MASKED_PROGRAMS
+        .get_or_init(|| RwLock::new(HashSet::new()))
+        .write()
+        .expect("masked program set lock poisoned");
+    for name in &names {
+        masked.insert(name.clone());
+    }
+    drop(masked);
+    MaskedPrograms(names)
+}
+
+/// Guard returned by [`mask_test_programs`]; lifts the mask on drop.
+#[cfg(any(test, feature = "test-support"))]
+#[doc(hidden)]
+pub struct MaskedPrograms(Vec<OsString>);
+
+#[cfg(any(test, feature = "test-support"))]
+impl Drop for MaskedPrograms {
+    fn drop(&mut self) {
+        if let Some(masked) = MASKED_PROGRAMS.get() {
+            let mut masked = masked.write().expect("masked program set lock poisoned");
+            for name in &self.0 {
+                masked.remove(name);
+            }
+        }
+    }
+}
+
+#[cfg(any(test, feature = "test-support"))]
+fn program_is_masked(program: &OsStr) -> bool {
+    MASKED_PROGRAMS.get().is_some_and(|masked| {
+        masked
+            .read()
+            .expect("masked program set lock poisoned")
+            .contains(program)
+    })
+}
+
+/// Resolve an executable when tests mask it or explicitly override `PATH`.
 /// Normal test-support builds keep the platform's native command lookup semantics.
 #[cfg(any(test, feature = "test-support"))]
 fn resolve_test_program(program: &OsStr) -> Option<std::path::PathBuf> {
     let program_path = std::path::Path::new(program);
     if program_path.components().count() != 1 {
         return None;
     }
+    if program_is_masked(program) {
+        return Some(std::path::Path::new(MASKED_PROGRAM_DIR).join(program));
+    }
     let search_path = test_override(OsStr::new("PATH"))??;
     std::env::split_paths(&search_path)
         .map(|dir| dir.join(program))
@@ -180,6 +256,66 @@ mod tests {
         assert!(var_os(KEY).is_none());
     }
 
+    /// Masking must make a program that *is* on `PATH` resolve to something that
+    /// cannot be spawned, and must lift cleanly — otherwise it becomes the same
+    /// process-global leak that replacing `PATH` was.
+    #[cfg(unix)]
+    #[test]
+    fn masking_hides_a_resolvable_program_and_lifts_on_drop() {
+        // Restore the effective PATH: `remove_test_var` masks the key outright,
+        // which would leave every later test in this binary with no PATH at all.
+        struct PathGuard(Option<OsString>);
+        impl Drop for PathGuard {
+            fn drop(&mut self) {
+                match self.0.take() {
+                    Some(previous) => set_test_var("PATH", previous),
+                    None => remove_test_var("PATH"),
+                }
+            }
+        }
+
+        // A name nothing else in this binary spawns, so the process-global mask
+        // cannot disturb a concurrently running test.
+        const PROGRAM: &str = "cortex-test-mask-probe";
+
+        let dir = tempfile::tempdir().unwrap();
+        let binary = dir.path().join(PROGRAM);
+        std::fs::write(&binary, "#!/bin/sh\nexit 0\n").unwrap();
+        use std::os::unix::fs::PermissionsExt;
+        std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o755)).unwrap();
+
+        // Prepend, never replace: a bare fixture PATH breaks unrelated spawns.
+        let mut search_path = vec![dir.path().to_path_buf()];
+        if let Some(existing) = var_os("PATH") {
+            search_path.extend(std::env::split_paths(&existing));
+        }
+        let _path = PathGuard(var_os("PATH"));
+        set_test_var("PATH", std::env::join_paths(search_path).unwrap());
+
+        assert_eq!(
+            resolve_test_program(OsStr::new(PROGRAM)).as_deref(),
+            Some(binary.as_path()),
+            "the fixture directory should resolve the program before masking"
+        );
+
+        {
+            let _masked = mask_test_programs([PROGRAM]);
+            let resolved = resolve_test_program(OsStr::new(PROGRAM))
+                .expect("a masked program still resolves, to a path that cannot exist");
+            assert!(
+                !resolved.exists(),
+                "masking must resolve to an absent path so the spawn fails with \
+                 NotFound, exactly as an uninstalled binary does"
+            );
+        }
+
+        assert_eq!(
+            resolve_test_program(OsStr::new(PROGRAM)).as_deref(),
+            Some(binary.as_path()),
+            "dropping the guard must lift the mask"
+        );
+    }
+
     #[test]
     fn unrelated_keys_do_not_interfere() {
         const LEFT: &str = "CORTEX_TEST_ENV_OVERLAY_LEFT";

src/heartbeat_agent_tests.rs

@@ -720,6 +720,7 @@ fn transcript_forward_env_does_not_gate_local_sessions_watch_service() {
 /// dead or permission-denied daemon — a CI container, for instance — reported a
 /// probe error instead. Both are the same fact about the host.
 #[tokio::test]
+#[serial]
 async fn container_probe_reports_unreachable_when_docker_ps_fails() {
     let dir = tempfile::tempdir().unwrap();
     let fake_docker = dir.path().join("docker");
@@ -733,7 +734,17 @@ async fn container_probe_reports_unreachable_when_docker_ps_fails() {
         use std::os::unix::fs::PermissionsExt;
         std::fs::set_permissions(&fake_docker, std::fs::Permissions::from_mode(0o755)).unwrap();
     }
-    crate::env::set_test_var("PATH", dir.path().to_str().unwrap());
+    // `PATH` overrides live in a process-global map that every subprocess spawn
+    // in this test binary resolves against, so this has to both prepend rather
+    // than replace and restore on drop. A bare, unrestored fixture directory
+    // outlives the test and leaves every later `sh`, `git`, or `docker` spawn
+    // in the binary resolving against a deleted temp dir. `cargo nextest` hides
+    // that behind one process per test; plain `cargo test` does not.
+    let mut search_path = vec![dir.path().to_path_buf()];
+    if let Some(existing) = crate::env::var_os("PATH") {
+        search_path.extend(std::env::split_paths(&existing));
+    }
+    let _path = EnvGuard::set("PATH", std::env::join_paths(search_path).unwrap());
 
     let output = LinuxContainerProbe.collect().await.expect(
         "a failing `docker ps` is an unreachable runtime, not a probe error — \

src/inventory/device_tests.rs

@@ -117,7 +117,20 @@ async fn collect_warns_when_optional_device_commands_are_missing() {
     let bin_dir = dir.path().join("bin");
     std::fs::create_dir_all(&bin_dir).unwrap();
     executable_file(&bin_dir.join("hostname"), "#!/bin/sh\nprintf '\\n'\n");
-    let _path_guard = EnvGuard::set("PATH", bin_dir.as_os_str());
+    // Mask the optional commands instead of replacing `PATH` with `bin_dir`.
+    // `PATH` overrides are process-global, so wiping it here also stopped every
+    // concurrently-running test in this binary from finding `sh`, `git`, and
+    // friends. Masking says what this test actually means — these four commands
+    // are not installed — and leaves every other program resolving normally.
+    // Safe to mask process-wide because `inventory::device` is the only caller
+    // of them and its tests are `#[serial]` with each other.
+    let _masked = crate::env::mask_test_programs(["uname", "ip", "ss", "df"]);
+    let path = format!(
+        "{}:{}",
+        bin_dir.display(),
+        crate::env::var("PATH").unwrap_or_default()
+    );
+    let _path_guard = EnvGuard::set("PATH", path);
 
     let output = collect(std::time::Duration::from_millis(50)).await;
 

src/runtime.rs

@@ -511,11 +511,14 @@ impl RuntimeCore {
         let timeline_rollup = self.spawn_timeline_rollup_task(token.clone());
         let optimize = self.spawn_optimize_task(token.clone());
         let file_tail = self.spawn_file_tail_task(token.clone());
+        // The projector's per-cycle progress receiver is a test observation
+        // seam; the runtime only needs the join handle for shutdown.
         let agent_observatory_projector = agent_observatory::spawn_projector(
             token.clone(),
             Arc::clone(&self.pool),
             self.config.agent_observatory.clone(),
-        );
+        )
+        .map(|(task, _progress)| task);
         let agent_observatory_git = agent_observatory::spawn_git_reconcile(
             token.clone(),
             Arc::clone(&self.pool),

src/runtime/agent_observatory.rs

@@ -16,6 +16,7 @@ use crate::scanner::local_hostname;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
+use tokio::sync::watch;
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 
@@ -37,14 +38,42 @@ fn projector_retry_delay_ms(attempt: usize) -> u64 {
     TRANSIENT_SQLITE_RETRY_DELAYS_MS[index]
 }
 
+/// Cumulative, monotonic progress of one projector instance.
+///
+/// A projector cycle has no wall-clock bound: it takes the process-wide SQLite
+/// write serialization lock (`db::pool::write_lock`) several times, so under
+/// contention — most visibly the parallel test suite, where every test's pool
+/// queues on that one lock — a single cycle can take tens of seconds. Anything
+/// that needs to know a cycle happened must therefore observe this signal
+/// rather than assume a cycle fits inside some interval.
+///
+/// Every field is cumulative and monotonic, so an observer can never miss a
+/// transition by sampling late — unlike the health row, whose `detail` describes
+/// only the cycle that wrote it and is overwritten by the next one.
+/// `health_records` counts only successfully persisted health rows, so it
+/// tracks the `attempts` counter inside the stored health JSON exactly.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(super) struct ProjectorProgress {
+    pub(super) cycles: u64,
+    pub(super) health_records: u64,
+    pub(super) projected: u64,
+    pub(super) oversized_first_rows: u64,
+}
+
+/// Spawn the projector, returning its join handle plus a receiver that reports
+/// [`ProjectorProgress`] after every completed cycle. Production drops the
+/// receiver; `watch::Sender::send_replace` never fails on a dropped receiver,
+/// so the signal costs one `watch` cell whether or not anyone is listening.
 pub(super) fn spawn_projector(
     token: CancellationToken,
     pool: Arc<DbPool>,
     config: AgentObservatoryConfig,
-) -> Option<JoinHandle<()>> {
+) -> Option<(JoinHandle<()>, watch::Receiver<ProjectorProgress>)> {
     config.enabled.then(|| {
         let mut wake = projection_wake_receiver();
-        tokio::spawn(async move {
+        let (progress_tx, progress_rx) = watch::channel(ProjectorProgress::default());
+        let task = tokio::spawn(async move {
+            let mut progress = ProjectorProgress::default();
             let mut consecutive_retry_safe_failures = 0usize;
             loop {
                 if token.is_cancelled() { break; }
@@ -158,7 +187,7 @@ pub(super) fn spawn_projector(
                     None
                 };
                 let status = if healthy { "ok" } else { "error" };
-                if let Err(error) = record_projection_health(
+                match record_projection_health(
                     &pool,
                     "projector",
                     status,
@@ -167,9 +196,20 @@ pub(super) fn spawn_projector(
                         retry_delay_ms.unwrap_or(0)
                     ),
                 ) {
-                    tracing::error!(error = %error, "Agent Observatory projector health write failed");
+                    Ok(()) => progress.health_records = progress.health_records.saturating_add(1),
+                    Err(error) => {
+                        tracing::error!(error = %error, "Agent Observatory projector health write failed");
+                    }
                 }
                 tracing::debug!(projected, "Agent Observatory projector cycle completed");
+                progress.cycles = progress.cycles.saturating_add(1);
+                progress.projected = progress.projected.saturating_add(projected as u64);
+                progress.oversized_first_rows = progress
+                    .oversized_first_rows
+                    .saturating_add(oversized_first_rows as u64);
+                // Published after the health row is committed so an observer
+                // woken by this signal always reads the row this cycle wrote.
+                progress_tx.send_replace(progress);
                 if healthy {
                     tokio::select! {
                         biased;
@@ -186,7 +226,8 @@ pub(super) fn spawn_projector(
                     }
                 }
             }
-        })
+        });
+        (task, progress_rx)
     })
 }