test(harness): parent cleanup, join gate, fault seams

crates/bashkit/tests/integration/threat_model_tests.rs

@@ -74,6 +74,22 @@ mod resource_exhaustion {
     /// 2s + reap 5s + completion 5s; every stage is finite.
     const JOIN_GRACE: Duration = Duration::from_secs(5);
 
+    /// Joins one reader handle inside the shared completion grace.
+    /// Returns true only when the handle was observed finished AND the
+    /// join succeeded: the single bounded-completion primitive used by
+    /// both the runner gate below and its deterministic proof test.
+    fn join_finished(handle: std::thread::JoinHandle<()>, end: Instant) -> bool {
+        let mut finished = handle.is_finished();
+        while !finished && Instant::now() < end {
+            std::thread::sleep(Duration::from_millis(5));
+            finished = handle.is_finished();
+        }
+        if !finished {
+            return false;
+        }
+        handle.join().is_ok()
+    }
+
     /// Reads one pipe up to the cap. Retention is truncated BEFORE the
     /// buffer can grow past it, so a hostile stream can never inflate the
     /// harness itself. `Interrupted` is retried; any other I/O error ends
@@ -164,10 +180,19 @@ mod resource_exhaustion {
         run
     }
 
+    fn run_raw_bounded(program: &str, args: &[&str], limit: Duration) -> BoundedRun {
+        run_raw_bounded_with(program, args, limit, LifecycleFaults::default())
+    }
+
     /// Runs an arbitrary program under the same fail-closed launcher (used
     /// by harness selftests so they prove the effective CPU/stack bounds
     /// instead of bypassing them). The caller supplies piped stdio.
-    fn run_raw_bounded(program: &str, args: &[&str], limit: Duration) -> BoundedRun {
+    fn run_raw_bounded_with(
+        program: &str,
+        args: &[&str],
+        limit: Duration,
+        faults: LifecycleFaults,
+    ) -> BoundedRun {
         let launcher = limit_launcher();
         let mut cmd = Command::new("bash");
         cmd.args(["-c", &launcher, "_"])
@@ -176,7 +201,7 @@ mod resource_exhaustion {
             .stdin(Stdio::null())
             .stdout(Stdio::piped())
             .stderr(Stdio::piped());
-        let mut run = run_command_bounded(cmd, limit);
+        let mut run = run_command_bounded_with(cmd, limit, faults);
         strip_launcher_note(&mut run);
         run
     }
@@ -331,22 +356,29 @@ mod resource_exhaustion {
                 }
             }
             if Instant::now() >= deadline {
+                // Verdict only: termination happens in common cleanup below,
+                // so every path (timeout, oversized, poll failure) funnels
+                // through one reachable kill guard. An unobservable child
+                // keeps its distinct verdict (incomplete, exitless, and no
+                // timeout label) instead of being mislabeled.
                 if status.is_none() && !oversized && !poll_failed {
-                    let _ = child.kill();
                     timed_out = true;
                 }
-                if status.is_none() && poll_failed {
-                    // Unobservable child: kill it like a timeout victim,
-                    // but the verdict stays distinct (incomplete below,
-                    // exitless, no timeout).
-                    let _ = child.kill();
-                }
                 break;
             }
             std::thread::sleep(Duration::from_millis(10));
         }
         // Bounded drain grace: an exited child with unretained pipes hits
         // EOF promptly; anything still missing is a retained pipe.
+        // Common cleanup termination: every loop exit (observed
+        // completion, deadline, oversized, poll failure) funnels through
+        // this one reachable kill guard. It is idempotent — already-dead
+        // children are untouched — and the live oversize kill inside the
+        // loop stays as the early producer stop. Verdicts were recorded
+        // pre-kill above and are never recomputed from it.
+        if status.is_none() {
+            let _ = child.kill();
+        }
         let grace_end = Instant::now() + DRAIN_GRACE;
         while (bufs[0].is_none() || bufs[1].is_none()) && Instant::now() < grace_end {
             match rx.recv_timeout(Duration::from_millis(50)) {
@@ -383,16 +415,7 @@ mod resource_exhaustion {
         let completion_end = Instant::now() + JOIN_GRACE;
         if bufs[0].is_some() && bufs[1].is_some() {
             for reader in readers {
-                let mut finished = reader.is_finished();
-                while !finished && Instant::now() < completion_end {
-                    std::thread::sleep(Duration::from_millis(5));
-                    finished = reader.is_finished();
-                }
-                if finished {
-                    if reader.join().is_err() {
-                        incomplete_readers = true;
-                    }
-                } else {
+                if !join_finished(reader, completion_end) {
                     incomplete_readers = true;
                 }
             }
@@ -406,11 +429,15 @@ mod resource_exhaustion {
         // stays exactly as recorded pre-kill and is never recomputed here.
         // Unavailable reap (or injected): the seam verdict flows into
         // the real parent status instead of a helper-only simulation.
+        // The OS cleanup always runs (no zombies by construction);
+        // reap_never only discards the observation for the verdict, never
+        // the cleanup itself. Do not invent exit/reap success from it.
+        let mut poll = || child.try_wait();
+        let really_reaped = reap_with_grace(&mut poll, REAP_GRACE);
         let reaped = if faults.reap_never {
             None
         } else {
-            let mut poll = || child.try_wait();
-            reap_with_grace(&mut poll, REAP_GRACE)
+            really_reaped
         };
         let status = match (status, reaped) {
             (Some(s), _) => Some(s),
@@ -629,65 +656,86 @@ mod resource_exhaustion {
 
     #[test]
     fn harness_join_gate_proves_paused_sender_unfinished() {
-        // A sender paused after send: deterministic proof that a channel
-        // receipt is not the termination contract. The sender blocks on
-        // `release` right after sending, so it is alive by construction
-        // when the parent observes the receipt — no timing luck involved.
-        // The thread body after release is a bare return (no blocking
-        // calls), so releasing it afterwards is a bounded cleanup path.
+        // A sender paused after send, driven through the ACTUAL
+        // `join_finished` gate the runner uses — not a parallel
+        // demonstration. The sender blocks on `release` right after
+        // sending, so it is alive by construction when the parent
+        // observes the receipt: no timing luck involved. The thread body
+        // after release is a bare return (no blocking calls), so the
+        // release below is a bounded cleanup path, watched by its own
+        // finite deadline.
         let (done_tx, done_rx) = std::sync::mpsc::channel::<()>();
         let (rel_tx, rel_rx) = std::sync::mpsc::channel::<()>();
         let handle = std::thread::spawn(move || {
             let _ = done_tx.send(());
             let _ = rel_rx.recv();
         });
-        done_rx.recv().expect("sender ran to the pause point");
+        done_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("sender ran to the pause point inside the readiness bound");
         assert!(
-            !handle.is_finished(),
-            "paused-after-send thread must read unfinished"
+            !join_finished(handle, Instant::now() + Duration::from_millis(100)),
+            "paused-after-send thread must fail the gate inside a short grace"
         );
-        // Release is the explicit cleanup path: the receiver fails, the
-        // thread returns, and the watchdog below proves it terminates.
-        let start = Instant::now();
+        // Release the negative-path thread at once: its receiver fails and
+        // it returns on its own (bare return, no blocking calls), so no
+        // detached thread outlives this test's scope.
         drop(rel_tx);
-        let mut finished = handle.is_finished();
-        while !finished && start.elapsed() < Duration::from_secs(5) {
-            std::thread::sleep(Duration::from_millis(5));
-            finished = handle.is_finished();
-        }
-        assert!(finished, "released thread must finish inside the watchdog");
+        // The release twin below proves the same gate primitive joins a
+        // released thread inside the watchdog.
+        let start = Instant::now();
+        // NOTE: the negative gate above consumed the handle; spawn the
+        // release twin whose only job is the bounded cleanup path.
+        let (done2_tx, done2_rx) = std::sync::mpsc::channel::<()>();
+        let (rel2_tx, rel2_rx) = std::sync::mpsc::channel::<()>();
+        let handle2 = std::thread::spawn(move || {
+            let _ = done2_tx.send(());
+            let _ = rel2_rx.recv();
+        });
+        done2_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("twin ran to the pause point inside the readiness bound");
+        drop(rel2_tx);
         assert!(
-            start.elapsed() < Duration::from_secs(5),
+            join_finished(handle2, Instant::now() + Duration::from_secs(5)),
+            "released thread must join inside the watchdog"
+        );
+        assert!(
+            start.elapsed() < Duration::from_secs(10),
             "cleanup path bounded"
         );
     }
 
     #[test]
     fn harness_parent_verdict_covers_injected_faults() {
-        // Poll failure through the REAL parent lifecycle: no panic, and a
-        // verdict distinct from timeout (incomplete, exitless, no timeout).
-        let mut c = Command::new("bash");
-        c.args(["-c", "exit 0"])
-            .stdin(Stdio::null())
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped());
-        let run = run_command_bounded_with(
-            c,
+        // Poll failure against a TRULY LIVE child through the REAL parent
+        // lifecycle: no panic, real termination in common cleanup, and a
+        // verdict distinct from timeout. The orphaned `sleep` is self-bound
+        // (30s); grandchildren cannot be reaped portably.
+        let start = Instant::now();
+        let run = run_raw_bounded_with(
+            "bash",
+            &["-c", "sleep 30"],
             Duration::from_secs(5),
             LifecycleFaults {
                 fail_poll: true,
                 ..Default::default()
             },
         );
-        // The reap still recovers the truthful exit of the already-dead
-        // child; the verdict stays distinct from a timeout (incomplete,
-        // exitless-timeout-absence) and from a clean run (incomplete).
-        assert_eq!(run.exit_code, Some(0), "reap recovers the true exit");
+        let elapsed = start.elapsed();
+        // The common cleanup really terminated the live child; the exit
+        // comes back signaled (no code), and the verdict stays distinct
+        // from a timeout (incomplete, exitless, no timeout label).
+        assert_eq!(run.exit_code, None, "killed live child has no exit code");
         assert!(run.incomplete_readers, "poll failure fails the run visibly");
         assert!(
             !run.timed_out,
             "poll failure must not be mislabeled as a timeout"
         );
+        assert!(
+            elapsed < Duration::from_secs(15),
+            "cleanup bounded without waiting for the 30s sleeper: {elapsed:?}"
+        );
         // Injected read failure through the REAL parent capture path:
         // partial capture with read_error set, never a clean EOF.
         let mut c = Command::new("bash");