Reap the bounded runner's leader only after its process group is retired

autoform_cli/skeleton.py

@@ -878,6 +878,8 @@ def _remember_descendants(
 ) -> None:
     """Retain handles for children that may outlive their immediate parent."""
 
+    if process.returncode is not None:
+        return  # a reaped leader's PID may belong to another process
     try:
         children = psutil.Process(process.pid).children(recursive=True)
     except (psutil.Error, OSError, SystemError):
@@ -952,6 +954,49 @@ def _process_group_is_alive(pid: int) -> bool:
     return True
 
 
+def _process_group_has_live_members(pid: int) -> bool:
+    """Whether the group ``pid`` leads holds a process that is not a zombie.
+
+    Its leader stays unreaped, and so a member, until the group is retired.
+    """
+
+    if not _process_group_is_alive(pid):
+        return False
+    try:
+        for candidate in psutil.process_iter(["pid", "status"]):
+            try:
+                if os.getpgid(candidate.info["pid"]) == pid and candidate.info["status"] != psutil.STATUS_ZOMBIE:
+                    return True
+            except (OSError, psutil.Error):
+                continue
+    except (OSError, psutil.Error, SystemError):
+        return True  # a table that cannot be read proves nothing gone
+    return False
+
+
+def _leader_has_exited(process: subprocess.Popen[bytes]) -> bool:
+    """Whether the command has exited, leaving it unreaped on POSIX.
+
+    ``Popen.poll`` would reap it, and a reaped leader's process group id can
+    pass to an unrelated group before the next ``killpg``.
+    """
+
+    if os.name != "posix" or process.returncode is not None:
+        return process.poll() is not None
+    try:
+        if hasattr(os, "waitid"):
+            # Unlike a zombie status, this waits for every thread of the leader.
+            return os.waitid(os.P_PID, process.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) is not None
+        return psutil.Process(process.pid).status() == psutil.STATUS_ZOMBIE
+    except psutil.ZombieProcess:
+        return True
+    except (ChildProcessError, psutil.NoSuchProcess):
+        # Reaped elsewhere: poll records it, so nothing signals the group again.
+        return process.poll() is not None
+    except (OSError, psutil.Error, SystemError):
+        return False
+
+
 def _remaining(deadline: float) -> float:
     return max(0.0, deadline - time.perf_counter())
 
@@ -969,7 +1014,11 @@ def _terminate_process_tree(
     deadline: float,
     token: str,
 ) -> None:
-    """Best-effort termination of a command and every descendant observed."""
+    """Best-effort termination of a command and every descendant observed.
+
+    The leader is reaped last, so the group is only signalled while its id
+    cannot belong to another group.
+    """
 
     started, available = time.perf_counter(), _remaining(deadline)
     for share, group_signal, method in ((0.4, "SIGTERM", "terminate"), (0.8, "SIGKILL", "kill")):
@@ -981,17 +1030,21 @@ def _terminate_process_tree(
             strict=False,
         )
         live = [child for child in descendants.values() if child.pid != process.pid and _process_is_alive(child)]
-        if os.name == "posix":
+        if os.name == "posix" and process.returncode is None:
             with contextlib.suppress(OSError):
                 os.killpg(process.pid, getattr(signal, group_signal))
-        for target in ([process] if process.poll() is None else []) + live:
+        # On POSIX killpg already reached the leader; Popen.terminate would poll, and so reap, it.
+        leader = [process] if os.name != "posix" and process.poll() is None else []
+        for target in leader + live:
             with contextlib.suppress(OSError, psutil.Error, SystemError):
                 getattr(target, method)()
         phase_deadline = started + available * share
-        with contextlib.suppress(OSError, subprocess.TimeoutExpired):
-            process.wait(timeout=_remaining(phase_deadline))
+        while not _leader_has_exited(process) and _remaining(phase_deadline):
+            time.sleep(min(0.01, _remaining(phase_deadline)))
         with contextlib.suppress(OSError, psutil.Error, SystemError):
             psutil.wait_procs(live, timeout=_remaining(phase_deadline))
+    with contextlib.suppress(OSError, subprocess.TimeoutExpired):
+        process.wait(timeout=_remaining(deadline))
 
 
 class _CommandSignalled(BaseException):
@@ -1158,17 +1211,23 @@ def check_output() -> None:
             reader = threading.Thread(target=drain, args=(name, stream), daemon=True)
             reader.start()
             readers.append(reader)
-        while process.poll() is None:
+        while not _leader_has_exited(process):
             _remember_descendants(process, descendants)
             check_output()
             remaining = deadline - time.monotonic()
             if remaining <= 0:
                 raise _CommandTimedOut([f"{context} timed out after {timeout:g} seconds"])
             overflow.wait(min(0.05, remaining))
         _remember_descendants(process, descendants)
+        # A reap recorded elsewhere frees the group id, so it is not checked again.
+        group_alive = process.returncode is None and _process_group_has_live_members(process.pid)
         _remember_tagged_processes(token, descendants, root_pid=process.pid)
-        if _process_group_is_alive(process.pid) or any(map(_process_is_alive, descendants.values())):
+        if group_alive or any(map(_process_is_alive, descendants.values())):
             raise SkeletonError([f"{context} left descendant processes running"])
+        # The group is retired. No signal may land between the reap and the recorded status.
+        guard.disarm()
+        process.wait()
+        guard.arm()
         cleanup_deadline = time.perf_counter() + _PROCESS_TERMINATION_GRACE
         if not _join_readers(readers, deadline=time.perf_counter() + _PROCESS_TERMINATION_GRACE * 0.2):
             raise SkeletonError([f"{context} left descendant processes holding its output pipes open"])

tests/test_skeleton.py

@@ -444,6 +444,98 @@ def test_bounded_command_rejects_a_successful_parent_with_a_live_descendant(
     _assert_dies(int(child_pid.read_text(encoding="utf-8")), "descendant", "successful parent exit")
 
 
+_OUTLIVED_BY_A_CHILD = (
+    "import subprocess, sys; "
+    "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'], "
+    "stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)"
+)
+
+
+@pytest.mark.skipif(os.name != "posix", reason="process groups are POSIX-specific")
+@pytest.mark.parametrize("sigchld", ["default", "ignored"])
+@pytest.mark.parametrize("waitid", [True, False], ids=["waitid", "zombie-status"])
+@pytest.mark.parametrize(
+    ("program", "refusal"),
+    [("pass", None), (_OUTLIVED_BY_A_CHILD, "descendant processes"), ("import time; time.sleep(30)", "timed out")],
+    ids=["exits", "outlived", "timed-out"],
+)
+def test_bounded_command_signals_its_group_only_while_the_leader_is_unreaped(
+    tmp_path: Path, monkeypatch, program: str, refusal: str | None, waitid: bool, sigchld: str
+) -> None:
+    # A reaped leader's process group id can pass to an unrelated group, which
+    # a later killpg would then signal. Python before 3.13 has no os.waitid on
+    # macOS, where psutil's zombie status stands in, and a launcher that ignores
+    # SIGCHLD has the kernel reap every child as it exits.
+    leaders: dict[int, subprocess.Popen[bytes]] = {}
+    popen, killpg = subprocess.Popen, os.killpg
+
+    def remember(*args, **kwargs):
+        process = popen(*args, **kwargs)
+        leaders[process.pid] = process
+        return process
+
+    def unreaped(pid: int) -> bool:
+        try:
+            os.kill(pid, 0)  # a zombie still answers; a reaped process does not
+        except ProcessLookupError:
+            return False
+        return True
+
+    def checked_killpg(pid: int, signal_number: int) -> None:
+        assert pid not in leaders or (leaders[pid].returncode is None and unreaped(pid)), (
+            "signalled a reaped leader's group"
+        )
+        killpg(pid, signal_number)
+
+    monkeypatch.setattr("autoform_cli.skeleton.subprocess.Popen", remember)
+    monkeypatch.setattr("autoform_cli.skeleton.os.killpg", checked_killpg)
+    if not waitid:
+        monkeypatch.delattr(os, "waitid", raising=False)
+    previous = signal.signal(signal.SIGCHLD, signal.SIG_IGN if sigchld == "ignored" else signal.SIG_DFL)
+    try:
+        if refusal is None:
+            assert _bounded(tmp_path, program).returncode == 0
+        else:
+            with pytest.raises(SkeletonError, match=refusal):
+                _bounded(tmp_path, program, **({"timeout": 1} if refusal == "timed out" else {}))
+    finally:
+        signal.signal(signal.SIGCHLD, previous)
+    assert all(leader.returncode is not None for leader in leaders.values())
+
+
+@pytest.mark.skipif(os.name != "posix", reason="process groups are POSIX-specific")
+def test_bounded_command_defers_a_signal_that_arrives_while_the_leader_is_reaped(tmp_path: Path, monkeypatch) -> None:
+    # A signal between waitpid's reap and Popen's recorded status must not reach
+    # cleanup while the leader still looks unreaped, or cleanup would signal its
+    # freed group id. The signal is handled once the reap is recorded.
+    reaped: set[int] = set()
+    waitpid, killpg = os.waitpid, os.killpg
+
+    def signalled_waitpid(pid: int, options: int) -> tuple[int, int]:
+        result = waitpid(pid, options)
+        if result[0]:
+            reaped.add(result[0])
+            signal.raise_signal(signal.SIGTERM)
+        return result
+
+    def checked_killpg(pid: int, signal_number: int) -> None:
+        assert pid not in reaped, "signalled a reaped leader's group"
+        killpg(pid, signal_number)
+
+    delivered: list[int] = []
+    previous = signal.signal(signal.SIGTERM, lambda signum, frame: delivered.append(signum))
+    try:
+        with monkeypatch.context() as patch:
+            patch.setattr("autoform_cli.skeleton.os.waitpid", signalled_waitpid)
+            patch.setattr("autoform_cli.skeleton.os.killpg", checked_killpg)
+            with pytest.raises(SkeletonError, match="interrupted by SIGTERM"):
+                _bounded(tmp_path, "pass")
+    finally:
+        signal.signal(signal.SIGTERM, previous)
+    assert reaped
+    assert delivered == [signal.SIGTERM]
+
+
 @pytest.mark.skipif(os.name != "posix", reason="detached-session assertion is POSIX-specific")
 def test_bounded_command_finds_a_descendant_that_escapes_its_process_group(
     tmp_path: Path,