perf(tests): bound Notes server descendant cleanup

scripts/verify_notes_browser_runtime.py

@@ -11,6 +11,38 @@ import time
 from urllib.request import urlopen
 
 ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+
+def _stop_owned_server(server, identity):
+    """Close only identity-bound descendants, including MCP's new sessions.
+
+    Capture before signalling the parent: an orphaned stderr holder can no
+    longer be attributed by ancestry. Never search by executable name or kill
+    the caller's process group.
+    """
+    from src import process_lifecycle as lifecycle
+
+    descendants = (lifecycle.bind_descendants([identity.pid])
+                   if identity.verdict() == lifecycle.OWNED else [])
+    if os.name == "nt":
+        if identity.verdict() == lifecycle.OWNED:
+            lifecycle.taskkill_tree(identity.pid)
+    else:
+        outcome = lifecycle.terminate_identities(
+            [*(entry.identity for entry in reversed(descendants)), identity],
+            steps=lifecycle.term_kill_steps(10),
+        )
+        if outcome.survivors or outcome.unverified:
+            raise RuntimeError(f"Disposable Notes cleanup could not verify exit: {outcome}")
+    try:
+        logs, _ = server.communicate(timeout=2)
+    except subprocess.TimeoutExpired as error:
+        # Do not turn an unidentified pipe holder into an unbounded wait.
+        server.stdout.close()
+        raise RuntimeError("Disposable Notes output pipe remained open after owned cleanup") from error
+    return logs
 
 
 def main():
@@ -36,7 +68,10 @@ def main():
         server = subprocess.Popen(
             [sys.executable, "-m", "uvicorn", "app:app", "--host", "127.0.0.1", "--port", str(port)],
             cwd=target, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
+            start_new_session=(os.name != "nt"),
         )
+        from src.process_lifecycle import ProcessIdentity
+        server_identity = ProcessIdentity.capture(server.pid)
         result = None
         try:
             origin = f"http://127.0.0.1:{port}"
@@ -62,12 +97,7 @@ def main():
                 print(result.stderr, file=sys.stderr)
             return result.returncode
         finally:
-            server.terminate()
-            try:
-                logs, _ = server.communicate(timeout=10)
-            except subprocess.TimeoutExpired:
-                server.kill()
-                logs, _ = server.communicate()
+            logs = _stop_owned_server(server, server_identity)
             if result is None or result.returncode:
                 print(logs[-8000:], file=sys.stderr)
 

tests/test_admin_listing_authorization.py

@@ -14,22 +14,12 @@ from routes.auth_routes import SESSION_COOKIE
 from tests.helpers.database import disposable_database
 
 
-@pytest.fixture(scope="module")
-def auth_configuration(tmp_path_factory):
-    """Hash real fixture passwords once; only immutable bytes are reused."""
-    path = tmp_path_factory.mktemp("admin-auth-seed") / "auth.json"
-    auth = AuthManager(str(path))
+@pytest.fixture
+def authenticated_app(tmp_path, monkeypatch):
+    monkeypatch.setenv("AUTH_ENABLED", "true")
+    auth = AuthManager(str(tmp_path / "auth.json"))
     for user, admin in (("member", False), ("admin", True)):
         assert auth.create_user(user, "disposable-test-password", is_admin=admin)
-    return path.read_bytes()
-
-
-@pytest.fixture
-def authenticated_app(tmp_path, monkeypatch, auth_configuration):
-    monkeypatch.setenv("AUTH_ENABLED", "true")
-    path = tmp_path / "auth.json"
-    path.write_bytes(auth_configuration)
-    auth = AuthManager(str(path))
     app = FastAPI()
     app.state.auth_manager = auth
 
@@ -143,17 +133,3 @@ def test_diagnostics_require_real_admin_before_sensitive_work(
         for operation in (stats, rag.get_stats, research.call_research_service,
                           extract_id, transcript, collect, read_log):
             operation.assert_not_called()
-
-
-@pytest.mark.parametrize("instance", [0, 1])
-def test_authenticated_fixture_has_private_users_sessions_and_file(authenticated_app, instance):
-    app, client_for = authenticated_app
-    auth = app.state.auth_manager
-    assert set(auth.users) == {"member", "admin"}
-    assert not auth._sessions
-    with client_for("member") as client:
-        assert auth.get_username_for_token(client.cookies.get(SESSION_COOKIE)) == "member"
-    assert auth.create_user("fixture-local", "disposable-test-password")
-    # Both parameter cases mutate their own file. Any shared manager, mutable
-    # users/sessions or written-back seed makes the next case fail, in any order.
-    assert "fixture-local" in auth.users

tests/test_notes_native_cleanup.py

@@ -0,0 +1,43 @@
+"""A detached Notes descendant must not hold output open or kill a bystander."""
+import os
+import runpy
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+from src import process_lifecycle as lifecycle
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX detached-session regression; Windows uses taskkill tree")
+def test_notes_cleanup_reaps_detached_pipe_holder_and_preserves_bystander():
+    stop = runpy.run_path(str(ROOT / "scripts/verify_notes_browser_runtime.py"))["_stop_owned_server"]
+    child = "import time; time.sleep(60)"
+    program = ("import subprocess,sys,time; "
+               f"p=subprocess.Popen([sys.executable,'-c',{child!r}],start_new_session=True); "
+               "print(p.pid,flush=True); time.sleep(60)")
+    with subprocess.Popen([sys.executable, "-c", child]) as bystander:
+        server = subprocess.Popen([sys.executable, "-c", program], stdout=subprocess.PIPE,
+                                  stderr=subprocess.STDOUT, text=True, start_new_session=True)
+        root_identity = lifecycle.ProcessIdentity.capture(server.pid)
+        detached = None
+        try:
+            child_pid = int(server.stdout.readline())
+            detached = lifecycle.ProcessIdentity.capture(child_pid)
+            assert os.getpgid(child_pid) == child_pid
+            stop(server, root_identity)
+            assert server.poll() is not None
+            assert detached.exited()
+            assert bystander.poll() is None, "cleanup touched an unrelated process"
+        finally:
+            # Test-owned fallback cleanup also runs when the negative control
+            # intentionally substitutes the old parent-only teardown.
+            targets = [identity for identity in (detached, root_identity) if identity]
+            lifecycle.terminate_identities(targets, steps=lifecycle.term_kill_steps(1))
+            server.wait(timeout=3)
+            if server.stdout:
+                server.stdout.close()
+            bystander.terminate()
+            bystander.wait(timeout=3)