fix(ci): isolate optional stack writers from test subprocess doubles

docs/development/testing-and-quality.md

@@ -731,6 +731,8 @@ Each triggered snapshot uses an isolated same-interpreter writer with a 100ms
 self-exit deadline, plus a 500ms parent communication timeout with kill/reap.
 This bounds pipe backpressure without changing shared stderr flags or leaving
 a persistently blocked writer after pytest exits; native process startup can add latency.
+The observer retains its own process constructor, so test replacements of public
+`subprocess.run` or `subprocess.Popen` cannot capture its diagnostic writer.
 Snapshots may be dropped on IO failure or backpressure. Default `0` starts no
 observer timer or writer process; pytest's own results and reports remain authoritative.
 The reusable Sonar workflow consumes that same run's XML and never reruns
@@ -758,6 +760,8 @@ JUnit、coverage 或必过门禁。归属与测量边界见
 每次触发栈输出时，使用同解释器的隔离 writer，100ms 自行退出；父进程另设
 500ms 通信期限，超时则终止并回收。管道背压不会改变共享 stderr 标志或留下
 持续阻塞的 writer；原生进程启动仍可能增加延迟。IO 失败或背压时可丢弃栈输出。
+观察器保留自己的进程构造入口，测试替换公共 `subprocess.run` 或 `subprocess.Popen`
+不会接管诊断 writer。
 默认 `0` 不启动观察 timer 或 writer 进程；pytest 原生结果和报告仍是权威。
 Sonar 只复用同一次 run 的 XML，不重复测试、不跨 run 取产物。缺少 token 仍成功跳过
 Sonar，测试 job 不接收 Sonar secret。触发范围取原有两套 workflow 的并集；纯前端

scripts/ci/pytest_timing.py

@@ -27,6 +27,9 @@
 
 PREFIX = "LOOPX_PYTEST_TIMING "
 _STACK_FD = pytest.StashKey[int]()
+# Tests may replace public subprocess helpers. The observer owns this transport;
+# subprocess.run would still resolve a test's replacement Popen at call time.
+_STACK_POPEN = subprocess.Popen
 _STACK_WRITE_PROGRAM = (
     "import os,sys,threading;"
     "deadline=threading.Timer(0.1,lambda:os._exit(0));"
@@ -62,13 +65,22 @@ def dump() -> None:
                 # A duplicate shares stderr's flags: never change them or
                 # write a potentially full pipe in this process. The isolated
                 # writer has its own exit deadline, even if pytest is killed;
-                # run's timeout also kills/reaps it before releasing this FD.
-                subprocess.run(
+                # The parent also kills/reaps it before releasing this FD.
+                with _STACK_POPEN(
                     [sys.executable, "-I", "-S", "-c", _STACK_WRITE_PROGRAM],
-                    input=("\n" + "\n".join(rows) + "\n").encode("utf-8", errors="backslashreplace"),
+                    stdin=subprocess.PIPE,
                     stdout=subprocess.DEVNULL, stderr=item.config.stash[_STACK_FD],
-                    timeout=0.5, check=False,
-                )
+                ) as writer:
+                    try:
+                        writer.communicate(
+                            ("\n" + "\n".join(rows) + "\n").encode("utf-8", errors="backslashreplace"),
+                            timeout=0.5,
+                        )
+                    finally:
+                        # Popen's context manager waits but does not kill on an
+                        # IO error. Stop our child before that unconditional reap.
+                        if writer.poll() is None:
+                            writer.kill()
             except (OSError, subprocess.TimeoutExpired):
                 # Optional diagnostics cannot replace pytest's result on IO failure.
                 pass

tests/test_ci_pytest_timing.py

@@ -165,38 +165,116 @@ def test_stall_snapshot_requires_original_descriptor_only_when_enabled(tmp_path,
 
 
 @pytest.mark.parametrize("workers", [0, 2])
-def test_stall_snapshot_output_failure_preserves_native_test_results(tmp_path, workers):
+@pytest.mark.parametrize("communication_failure", [False, pytest.param(
+    True, marks=pytest.mark.skipif(os.name == "nt", reason="POSIX process-group cleanup of a live fault child"),
+)], ids=["spawn", "communication"])
+def test_stall_snapshot_output_failure_preserves_native_test_results(tmp_path, workers, communication_failure):
     (tmp_path / "conftest.py").write_text(
-        "import os, subprocess, sys\nfrom pathlib import Path\n"
-        "original_run = subprocess.run\n"
+        "import json, os, sys\nfrom pathlib import Path\n"
+        "from scripts.ci import pytest_timing as observer\n"
+        "original_popen = observer._STACK_POPEN\nwriters = []\n"
         "def broken_snapshot(command, *args, **kwargs):\n"
         "    if command[:3] == [sys.executable, '-I', '-S']:\n"
         "        Path('attempt-' + str(os.getpid())).touch()\n"
+        f"        if {communication_failure}:\n"
+        "            process = original_popen([*command[:-1], 'import time; time.sleep(30)'], *args, **kwargs)\n"
+        "            writers.append(process)\n"
+        "            def broken_communicate(*args, **kwargs): raise OSError('diagnostic pipe unavailable')\n"
+        "            process.communicate = broken_communicate\n"
+        "            return process\n"
         "        raise OSError('diagnostic output unavailable')\n"
-        "    return original_run(command, *args, **kwargs)\n"
-        "def pytest_configure(config): subprocess.run = broken_snapshot\n",
+        "    return original_popen(command, *args, **kwargs)\n"
+        "def pytest_configure(config): observer._STACK_POPEN = broken_snapshot\n"
+        "def pytest_sessionfinish(session):\n"
+        "    assert all(process.poll() is not None for process in writers)\n"
+        "    Path('writers-' + str(os.getpid()) + '.json').write_text(json.dumps([p.returncode for p in writers]))\n",
         encoding="utf-8")
     (tmp_path / "test_subject.py").write_text(
         "import time\n"
-        "def test_pass(): time.sleep(0.03)\n"
+        "def test_pass(): time.sleep(0.1)\n"
         "def test_fail():\n"
-        "    time.sleep(0.03)\n"
+        "    time.sleep(0.1)\n"
         "    assert False\n", encoding="utf-8")
     log = tmp_path / "io-failure.log"
     with log.open("w") as output:
-        result = subprocess.run(
+        process = subprocess.Popen(
             [sys.executable, "-m", "pytest", "-q", "-n", str(workers),
              "-p", "scripts.ci.pytest_timing", "--ci-stall-after=0.01"],
             cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT,
-            timeout=30, check=False)
-    assert result.returncode == 1
+            start_new_session=os.name != "nt")
+        try:
+            assert process.wait(timeout=5) == 1
+        finally:
+            if process.poll() is None:
+                if os.name == "nt":
+                    process.kill()
+                else:
+                    os.killpg(process.pid, signal.SIGKILL)
+            process.wait(timeout=10)
     assert list(tmp_path.glob("attempt-*"))
+    writers = [code for path in tmp_path.glob("writers-*.json") for code in json.loads(path.read_text())]
+    assert len(writers) == (2 if communication_failure else 0)
+    assert all(code < 0 for code in writers)
     assert "LOOPX_PYTEST_STACK " not in log.read_text()
     report = read_log(log)
     assert report["execution_complete"] and report["pytest_exit_code"] == 1
     assert report["selected_cases"] == report["finished_cases"] == 2
 
 
+@pytest.mark.parametrize("workers", [0, 2])
+@pytest.mark.parametrize("target", ["run", "Popen"])
+@pytest.mark.parametrize("enabled", [False, True])
+@pytest.mark.skipif(os.name == "nt", reason="POSIX process-group cleanup of the failing baseline")
+def test_test_owned_subprocess_doubles_cannot_capture_diagnostic_writers(tmp_path, workers, target, enabled):
+    (tmp_path / "conftest.py").write_text(
+        "import subprocess, sys, threading, pytest\nfrom pathlib import Path\n"
+        "from scripts.ci import pytest_timing as observer\n"
+        "original_popen = subprocess.Popen\nwriters = []\n"
+        "def tracked_writer(*args, **kwargs):\n"
+        "    process = original_popen(*args, **kwargs)\n"
+        "    writers.append(process)\n"
+        "    return process\n"
+        "observer._STACK_POPEN = tracked_writer\n"
+        f"original = subprocess.{target}\n"
+        "def test_double(command, *args, **kwargs):\n"
+        "    if command[:3] == [sys.executable, '-I', '-S']:\n"
+        "        Path('shared-subprocess-entered').touch()\n"
+        "        threading.Event().wait(30)\n"
+        "        raise AssertionError('test double must not own diagnostic transport')\n"
+        "    return original(command, *args, **kwargs)\n"
+        f"def pytest_configure(config): subprocess.{target} = test_double\n"
+        "@pytest.hookimpl(wrapper=True, tryfirst=True)\n"
+        "def pytest_runtest_protocol(item):\n"
+        "    before = set(threading.enumerate())\n"
+        "    result = yield\n"
+        "    assert not any(isinstance(t, threading.Timer) and t not in before\n"
+        "                   for t in threading.enumerate())\n"
+        "    assert all(p.poll() is not None for p in writers)\n"
+        "    return result\n", encoding="utf-8")
+    (tmp_path / "test_subject.py").write_text(
+        "import time\n"
+        "def test_pass(): time.sleep(0.1)\n"
+        "def test_fail():\n    time.sleep(0.1)\n    assert False\n", encoding="utf-8")
+    log = tmp_path / "test-double.log"
+    with log.open("w") as output:
+        process = subprocess.Popen(
+            [sys.executable, "-m", "pytest", "-q", "-n", str(workers),
+             "-p", "scripts.ci.pytest_timing", "--ci-stall-after=" + str(0.01 if enabled else 0)],
+            cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT,
+            start_new_session=True)
+        try:
+            assert process.wait(timeout=5) == 1
+        finally:
+            if process.poll() is None:
+                os.killpg(process.pid, signal.SIGKILL)
+            process.wait(timeout=10)
+    assert not (tmp_path / "shared-subprocess-entered").exists()
+    assert ("LOOPX_PYTEST_STACK " in log.read_text()) == enabled
+    report = read_log(log)
+    assert report["execution_complete"] and report["pytest_exit_code"] == 1
+    assert report["selected_cases"] == report["finished_cases"] == 2
+
+
 @pytest.mark.parametrize("workers", [0, 2])
 def test_completed_tests_cancel_pending_stall_snapshot(tmp_path, workers):
     (tmp_path / "test_subject.py").write_text("def test_quick(): pass\n", encoding="utf-8")
@@ -228,8 +306,9 @@ def unread_stderr_pipe():
 
 def track_stack_writers(tmp_path, *, child_deadline=True):
     (tmp_path / "conftest.py").write_text(
-        "import json, subprocess, sys\nfrom pathlib import Path\n"
-        "original_popen = subprocess.Popen\nwriters = []\n"
+        "import json, sys\nfrom pathlib import Path\n"
+        "from scripts.ci import pytest_timing as observer\n"
+        "original_popen = observer._STACK_POPEN\nwriters = []\n"
         "def tracked_popen(command, *args, **kwargs):\n"
         "    diagnostic = command[:3] == [sys.executable, '-I', '-S']\n"
         f"    if diagnostic and {not child_deadline}:\n"
@@ -239,7 +318,7 @@ def track_stack_writers(tmp_path, *, child_deadline=True):
         "        writers.append(process)\n"
         "        Path('writer.pid').write_text(str(process.pid))\n"
         "    return process\n"
-        "subprocess.Popen = tracked_popen\n"
+        "observer._STACK_POPEN = tracked_popen\n"
         "def pytest_sessionfinish(session):\n"
         "    assert all(process.poll() is not None for process in writers)\n"
         "    Path('writers.json').write_text(json.dumps([p.returncode for p in writers]))\n",