fix(runtime): require process exit before namespace disposal

docs/development/testing-and-quality.md

@@ -431,6 +431,18 @@ it differs from the live server before dispatching expensive work. Stop only
 that isolated runtime after its requests settle, retaining the same temporary
 environment for shutdown; never restart a live server as test cleanup.
 
+For namespace disposal, call `restart_effect_runtime(wait_for_exit=True)` and
+require `stopped` before removing files. A closed listener or a zombie process
+group leader does not prove that all runtime threads have exited. A PID that is
+still present returns `shutdown_pending`; retain the namespace and retry after
+the process exits and its parent reaps it. The ordinary restart action still
+uses serving-locator retirement.
+
+删除隔离目录前，调用 `restart_effect_runtime(wait_for_exit=True)` 并确认返回
+`stopped`。端口关闭或进程组长成为僵尸进程，都不能证明其他线程已经退出。
+PID 仍存在时返回 `shutdown_pending`；保留目录，等进程退出并由父进程回收后再试。
+普通重启操作仍按服务 locator 的退役状态判断。
+
 Automatic test cleanup must verify namespace ownership as well as PID/token
 identity. An ordinary copy of a valid locator inside a test directory is not
 permission to stop its original server. The private authenticated `runtime.ping`

loopx/control_plane/effect_runtime.py

@@ -771,22 +771,7 @@ def _stop_effect_runtime(
             if not _pid_is_alive(pid):
                 stopped = True
                 break
-            if os.name == "posix":
-                # A non-child zombie cannot write, but kill(pid, 0) still sees
-                # it. Managed runtimes start in their own session/process group.
-                from ..extensions.process_runtime import _posix_owned_group_has_exited
-                try:
-                    stopped = _posix_owned_group_has_exited(
-                        pid, max(.001, deadline - time.monotonic()),
-                    )
-                except (OSError, subprocess.TimeoutExpired, UnicodeError, RuntimeError):
-                    stopped = False
-                if stopped:
-                    # The child may have exited during the group snapshot,
-                    # after the first nonblocking reap. Retire its wait status
-                    # before returning successful namespace cleanup.
-                    _reap_exited_runtime_child(info)
-                    break
+            # A zombie leader can still have threads writing into the namespace.
             time.sleep(.05)
             continue
         published, published_token = _serving_token(info_path)

tests/control_plane/test_effect_runtime_integration.py

@@ -33,16 +33,17 @@
 
 
 @pytest.mark.parametrize("replacement", [False, True])
+@pytest.mark.parametrize("owns_group", [True, False] if os.name == "posix" else [False])
 def test_fixture_shutdown_waits_for_process_exit_after_locator_turnover(
-    tmp_path, monkeypatch, replacement,
+    tmp_path, monkeypatch, replacement, owns_group,
 ):
     locator = tmp_path / "runtime.json"
     if replacement:
         locator.write_text(json.dumps({"token": "replacement"}))
     monkeypatch.setattr(effect_runtime, "_request_with_info", lambda *a, **kw: {})
     with subprocess.Popen(
         [sys.executable, "-c", "import time; print('ready',flush=True); time.sleep(60)"],
-        stdout=subprocess.PIPE, text=True, start_new_session=os.name == "posix",
+        stdout=subprocess.PIPE, text=True, start_new_session=owns_group,
     ) as process:
         try:
             assert process.stdout.readline().strip() == "ready"
@@ -69,6 +70,50 @@ def test_fixture_shutdown_waits_for_process_exit_after_locator_turnover(
                 process.wait()
 
 
+@pytest.mark.skipif(sys.platform != "linux", reason="Linux thread-group leader exit")
+def test_fixture_shutdown_waits_for_writer_after_thread_group_leader_exits(tmp_path, monkeypatch):
+    witness = tmp_path / "written-after-stop"
+    monkeypatch.setattr(effect_runtime, "_request_with_info", lambda *a, **kw: {})
+    source = """
+import ctypes, os, pathlib, sys, threading
+def write_after_release():
+    print('ready', flush=True)
+    sys.stdin.readline()
+    pathlib.Path(sys.argv[1]).write_text('worker can still write')
+    os._exit(0)
+threading.Thread(target=write_after_release).start()
+ctypes.CDLL(None).pthread_exit(None)
+"""
+    with subprocess.Popen(
+        [sys.executable, "-c", source, str(witness)],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, start_new_session=True,
+    ) as process:
+        try:
+            assert process.stdout.readline().strip() == "ready"
+            state = Path(f"/proc/{process.pid}/stat")
+            deadline = time.monotonic() + 5
+            while state.read_text().rsplit(")", 1)[1].split()[0] != "Z":
+                assert time.monotonic() < deadline, "thread-group leader did not exit"
+                time.sleep(.01)
+            info = {"pid": process.pid, "token": "original"}
+            stopped = effect_runtime._stop_effect_runtime(
+                tmp_path / "retired-locator", info, timeout=.1, wait_for_process_exit=True,
+            )
+            assert stopped["status"] == "shutdown_pending"
+            assert effect_runtime._pid_is_alive(process.pid)
+            process.stdin.write("release\n")
+            process.stdin.flush()
+            process.wait(timeout=5)
+            assert witness.read_text() == "worker can still write"
+            assert effect_runtime._stop_effect_runtime(
+                tmp_path / "retired-locator", info, timeout=.1, wait_for_process_exit=True,
+            )["status"] == "stopped"
+        finally:
+            if process.poll() is None:
+                process.kill()
+                process.wait(timeout=5)
+
+
 def _effect_id(label: str) -> str:
     return f"fixture-goal:fixture-agent:{_TODO_ID}:{label}"
 

tests/control_plane/test_effect_runtime_restart.py

@@ -354,42 +354,30 @@ def unavailable(pid, timeout):
 
 
 @pytest.mark.skipif(os.name != "posix" or not hasattr(os, "WNOWAIT"), reason="POSIX child exit observation")
-def test_namespace_shutdown_reaps_child_exiting_during_group_observation(tmp_path, monkeypatch):
-    from loopx.extensions import process_runtime
-
+def test_namespace_shutdown_reaps_exited_child_before_disposal(tmp_path, monkeypatch):
     child = subprocess.Popen(
         [sys.executable, "-c", "import sys; sys.stdin.read()"],
         stdin=subprocess.PIPE, start_new_session=True,
     )
     info = {"pid": child.pid, "token": "serving"}
-    observe_group = process_runtime._posix_owned_group_has_exited
-    requests, observations = [], []
+    requests = []
 
     def shutdown(_info, **kwargs):
         requests.append(kwargs["method"])
-        # Keep the child live through the first nonblocking reap. It exits
-        # during the following real process-group observation instead.
-        return {}
-
-    def exited_during_observation(pid, timeout):
-        assert pid == child.pid and effect_runtime._pid_is_alive(pid)
+        assert effect_runtime._pid_is_alive(child.pid)
         child.stdin.close()
-        # Observe the real exit without reaping it. kill(pid, 0) still sees
-        # this zombie, while the group observer correctly sees no writer.
-        os.waitid(os.P_PID, pid, os.WEXITED | os.WNOWAIT)
-        exited = observe_group(pid, timeout)
-        observations.append(exited)
-        assert exited and effect_runtime._pid_is_alive(pid)
-        return exited
+        # Wait for the shutdown response's real exit without reaping the child.
+        os.waitid(os.P_PID, child.pid, os.WEXITED | os.WNOWAIT)
+        assert effect_runtime._pid_is_alive(child.pid)
+        return {}
 
     monkeypatch.setattr(effect_runtime, "_request_with_info", shutdown)
-    monkeypatch.setattr(process_runtime, "_posix_owned_group_has_exited", exited_during_observation)
     try:
         result = effect_runtime._stop_effect_runtime(
             tmp_path / "runtime.json", info, timeout=5, wait_for_process_exit=True,
         )
         assert result["stopped"] and result["status"] == "stopped"
-        assert requests == ["runtime.shutdown"] and observations == [True]
+        assert requests == ["runtime.shutdown"]
         assert not effect_runtime._pid_is_alive(child.pid), "Exited child must be reaped before cleanup returns"
         with pytest.raises(ChildProcessError):
             os.waitpid(child.pid, os.WNOHANG)