test: give the Windows descendant probe its own budget

scripts/run-test-wave.py

@@ -28,6 +28,20 @@
 SLOW_SUITES = frozenset(("incremental", "store_arch", "daemon_runtime"))
 POLL_SECONDS = 0.05
 
+# WHY: the Windows descendant probe below is a cold `powershell.exe` + CIM
+# start. On a GitHub Windows runner that routinely costs seconds -- interpreter
+# start-up, module autoload, CIM service warm-up -- and that cost is unrelated
+# to the state of the tree being proven. --kill-grace bounds how long a
+# *process* may resist termination and CI passes 1s, so timing the probe with
+# it made the proof a function of interpreter latency instead of the tree: a
+# cold start blew the 1s budget, TimeoutExpired became "assume the worst", and
+# an already-clean shard exited 2. This is a stable-state budget, not a race
+# tune -- the answer does not change with waiting, the budget only has to cover
+# a cold start, and a probe that still cannot finish is reported as an
+# unfinished probe rather than as a leaked tree.
+WINDOWS_DESCENDANT_PROBE_SECONDS = 15
+WINDOWS_DESCENDANT_PROBE_ATTEMPTS = 2
+
 
 @dataclass
 class ActiveSuite:
@@ -123,37 +137,54 @@ def start_suite(
     )
 
 
-def windows_descendants(pid: int, timeout: int) -> bool:
-    """True if any live process still claims `pid` as its parent.
+def windows_tree_cleanup_blocker(pid: int) -> str | None:
+    """Why `pid`'s tree cannot be called clean, or None when it provably is.
 
     Used only when the suite leader has already exited: `taskkill /T` cannot
     walk a tree from a dead PID, so cleanup is proven by asking whether anything
     is still parented to it. One level deep on purpose -- Windows does not
     reparent orphans, so a grandchild keeps pointing at its own (dead) parent
     and would not be found here. That is a weaker proof than taskkill /T, which
     is why it is reserved for the case where the strong proof is impossible.
+
+    Fail-closed: a probe that times out, cannot start, or reports failure is
+    never read as absence. The reason names WHICH of the two happened -- a probe
+    that did not finish, or a counted set of live descendants -- because those
+    are different defects and used to be reported with the same sentence.
     """
-    try:
-        completed = subprocess.run(
-            [
-                "powershell.exe",
-                "-NoProfile",
-                "-NonInteractive",
-                "-Command",
-                "@(Get-CimInstance Win32_Process -Filter "
-                f"'ParentProcessId={pid}').Count",
-            ],
-            check=False,
-            stdin=subprocess.DEVNULL,
-            capture_output=True,
-            text=True,
-            timeout=timeout,
-        )
-    except (OSError, subprocess.TimeoutExpired):
-        return True  # cannot prove absence -> assume the worst
-    if completed.returncode != 0:
-        return True
-    return (completed.stdout or "").strip() not in ("0", "")
+    unproven = "descendant probe did not run"
+    for _ in range(WINDOWS_DESCENDANT_PROBE_ATTEMPTS):
+        try:
+            completed = subprocess.run(
+                [
+                    "powershell.exe",
+                    "-NoProfile",
+                    "-NonInteractive",
+                    "-Command",
+                    "@(Get-CimInstance Win32_Process -Filter "
+                    f"'ParentProcessId={pid}').Count",
+                ],
+                check=False,
+                stdin=subprocess.DEVNULL,
+                capture_output=True,
+                text=True,
+                timeout=WINDOWS_DESCENDANT_PROBE_SECONDS,
+            )
+        except subprocess.TimeoutExpired:
+            unproven = (
+                "descendant probe could not complete in "
+                f"{WINDOWS_DESCENDANT_PROBE_SECONDS}s"
+            )
+            continue
+        except OSError as exc:
+            return f"descendant probe could not run: {exc}"
+        if completed.returncode != 0:
+            return f"descendant probe failed (rc={completed.returncode})"
+        count = (completed.stdout or "").strip()
+        if count in ("0", ""):
+            return None
+        return f"{count} live descendant(s)"
+    return unproven
 
 
 def terminate_process_tree(active: ActiveSuite, kill_grace: int) -> None:
@@ -167,9 +198,11 @@ def terminate_process_tree(active: ActiveSuite, kill_grace: int) -> None:
             # how a deliberately-hanging fixture suite reddened a release run.
             # taskkill /T cannot walk a tree from a dead PID, so prove cleanup
             # the only way still available -- nothing is parented to it.
-            if windows_descendants(process.pid, kill_grace):
+            blocker = windows_tree_cleanup_blocker(process.pid)
+            if blocker is not None:
                 raise RuntimeError(
-                    f"suite {active.name!r} leader exited leaving live descendants"
+                    f"suite {active.name!r} leader exited and tree cleanup "
+                    f"could not be proven: {blocker}"
                 )
             return
         try:

tests/test_parallel_harness_contract.sh

@@ -30,6 +30,99 @@ if ! grep -Fq 'run-test-wave.py' "$driver"; then
     exit 1
 fi
 
+# The Windows descendant proof must not be timed by --kill-grace. That argument
+# bounds how long a *process* may resist termination (this file runs the
+# scheduler with 1s); the probe is a cold PowerShell + CIM start that routinely
+# costs seconds on a runner. Binding one to the other made the verdict a
+# function of interpreter latency: a slow start became "assume the worst" and
+# reddened an already-clean shard. Asserted structurally -- no sleeps, no timing
+# thresholds -- so the contract stays deterministic on every platform.
+# (Command substitution, not `| grep -q`: under pipefail an early-exiting
+# reader can hand the writer EPIPE and turn a satisfied match into status 141.)
+probe_sites=$(grep -n 'windows_tree_cleanup_blocker(' "$scheduler" || true)
+if [[ "$probe_sites" == *kill_grace* ]]; then
+    echo "FAIL: the Windows descendant probe is still timed by --kill-grace" >&2
+    exit 1
+fi
+
+python3 - "$scheduler" <<'PROBE'
+from __future__ import annotations
+
+import importlib.util
+import subprocess
+import sys
+
+
+spec = importlib.util.spec_from_file_location("cbm_run_test_wave", sys.argv[1])
+module = importlib.util.module_from_spec(spec)
+# @dataclass resolves its own module out of sys.modules; register before exec.
+sys.modules[spec.name] = module
+spec.loader.exec_module(module)
+
+budget = getattr(module, "WINDOWS_DESCENDANT_PROBE_SECONDS", None)
+if not isinstance(budget, int) or budget < 15:
+    raise SystemExit(
+        "FAIL: the descendant probe has no independent budget "
+        f"(WINDOWS_DESCENDANT_PROBE_SECONDS={budget!r})"
+    )
+
+probe = module.windows_tree_cleanup_blocker
+original_run = subprocess.run
+observed: list[object] = []
+
+
+def timing_out(*args: object, **kwargs: object) -> object:
+    observed.append(kwargs.get("timeout"))
+    raise subprocess.TimeoutExpired(cmd="probe", timeout=kwargs.get("timeout"))
+
+
+class _Completed:
+    def __init__(self, stdout: str) -> None:
+        self.returncode = 0
+        self.stdout = stdout
+
+
+try:
+    subprocess.run = timing_out
+    timed_out_reason = probe(4321)
+    subprocess.run = lambda *a, **k: _Completed("3\n")
+    live_reason = probe(4321)
+    subprocess.run = lambda *a, **k: _Completed("0\n")
+    clean_reason = probe(4321)
+finally:
+    subprocess.run = original_run
+
+if observed != [budget] * len(observed):
+    raise SystemExit(
+        "FAIL: the descendant probe is not bounded by its own budget "
+        f"(timeouts={observed})"
+    )
+if len(observed) < 2:
+    raise SystemExit(
+        "FAIL: the descendant probe does not retry a timed-out probe "
+        f"(attempts={len(observed)})"
+    )
+if timed_out_reason is None or live_reason is None:
+    raise SystemExit(
+        "FAIL: the descendant probe stopped failing closed "
+        f"(timed_out={timed_out_reason!r}, live={live_reason!r})"
+    )
+if clean_reason is not None:
+    raise SystemExit(f"FAIL: a clean tree was not proven clean ({clean_reason!r})")
+if "could not complete" not in timed_out_reason:
+    raise SystemExit(
+        f"FAIL: an unfinished probe is not named as one ({timed_out_reason!r})"
+    )
+if "3 live descendant" not in live_reason:
+    raise SystemExit(
+        f"FAIL: proven descendants are not reported with their count ({live_reason!r})"
+    )
+if timed_out_reason == live_reason:
+    raise SystemExit(
+        "FAIL: an unfinished probe and a leaked tree are reported identically"
+    )
+PROBE
+
 cat >"$fixture/fake_runner.py" <<'PY'
 from __future__ import annotations
 
@@ -370,7 +463,11 @@ try:
             raise SystemExit("FAIL: scheduler did not observe the forced leader exit")
         time.sleep(0.02)
     release.write_text("release\n", encoding="utf-8")
-    stdout, stderr = process.communicate(timeout=8)
+    # Generous on purpose: the scheduler's refusal is the asserted state, and
+    # on Windows it now spends up to the descendant-probe budget (twice --
+    # once in the wave loop, once in the cleanup pass) before refusing. This
+    # bound only has to exceed that worst case; it never decides the verdict.
+    stdout, stderr = process.communicate(timeout=120)
 
     if os.name == "nt":
         # Assert the PROPERTY, not the wording. This used to require the phrase