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_editor_photoshop_features_more4.py

@@ -93,7 +93,7 @@ def test_custom_filter_reuses_convolve_kernel():
     assert "convolve(readSource(snap), k, divisor, bias, 1)" in EFFECTS
 
 
-def test_blur_and_custom_filter_aliases_registered():
+def test_new_filter_aliases_registered():
     assert "'box-blur': 'blur-box'" in EDITOR
     assert "'surface-blur': 'blur-surface'" in EDITOR
     assert "custom: 'other-custom'" in EDITOR

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)

tests/test_procfs_scan_guard.py

@@ -12,8 +12,6 @@ exercising every procfs-touching code path on both a Linux and a non-Linux
 host. ``test_cookbook_stop_without_procfs.py`` covers the behaviour itself.
 """
 import ast
-
-import pytest
 from pathlib import Path
 
 REPO_ROOT = Path(__file__).resolve().parent.parent
@@ -77,112 +75,37 @@ def _scan_sites(tree: ast.AST) -> list[ast.Call]:
     return sites
 
 
-def _condition_requires_procfs(node: ast.AST, truth: bool) -> bool:
-    """Whether this Boolean outcome proves a direct procfs check succeeded.
-
-    An ignored call, another function's check, or a check on only one possible
-    OR branch provides no proof. Keep unknown expressions conservative.
-    """
-    if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not):
-        return _condition_requires_procfs(node.operand, not truth)
-    if isinstance(node, ast.BoolOp):
-        facts = [_condition_requires_procfs(value, truth) for value in node.values]
-        all_values_required = isinstance(node.op, ast.And) == truth
-        return any(facts) if all_values_required else all(facts)
-    if not isinstance(node, ast.Call):
-        return False
-    func = node.func
-    name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
-    subject_is_procfs = (
-        name == "has_procfs"
-        or (isinstance(func, ast.Attribute) and _is_procfs_root(func.value))
-        or any(_is_procfs_root(arg) for arg in node.args)
-    )
-    return truth and name in _GUARD_FUNCS and subject_is_procfs
+def _guard_lines(tree: ast.AST) -> list[int]:
+    """Line numbers of calls that test whether procfs is present."""
+    lines = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
+        if name not in _GUARD_FUNCS:
+            continue
+        subject_is_procfs = (
+            name in {"has_procfs", "is_wsl"}
+            or (isinstance(func, ast.Attribute) and _is_procfs_root(func.value))
+            or any(_is_procfs_root(arg) for arg in node.args)
+        )
+        if subject_is_procfs:
+            lines.append(node.lineno)
+    return lines
 
 
-def _guarded_scan_ids(tree: ast.AST) -> set[int]:
-    """Follow conditional/early-exit paths instead of comparing line numbers.
-
-    Each function starts with no procfs assumption. Facts from a loop or try
-    body cannot escape to code which may run without that body completing.
-    This intentionally requires direct checks; arbitrary aliases/call graphs
-    are not a proof that procfs exists at the enumeration site.
-    """
-    scans = {id(site) for site in _scan_sites(tree)}
-    guarded = set()
-
-    def expression(node, present):
-        if node is None:
-            return
-        if id(node) in scans and present:
-            guarded.add(id(node))
-        if isinstance(node, ast.BoolOp):
-            for value in node.values:
-                expression(value, present)
-                # Later AND operands run only after a true prefix; later OR
-                # operands run only after a false prefix.
-                present = present or _condition_requires_procfs(value, isinstance(node.op, ast.And))
-        elif isinstance(node, ast.IfExp):
-            expression(node.test, present)
-            expression(node.body, present or _condition_requires_procfs(node.test, True))
-            expression(node.orelse, present or _condition_requires_procfs(node.test, False))
-        elif isinstance(node, (ast.ListComp, ast.SetComp, ast.GeneratorExp, ast.DictComp)):
-            for generator in node.generators:
-                expression(generator.iter, present)
-                for condition in generator.ifs:
-                    expression(condition, present)
-                    present = present or _condition_requires_procfs(condition, True)
-            for attr in ("elt", "key", "value"):
-                expression(getattr(node, attr, None), present)
-        elif isinstance(node, ast.Lambda):
-            expression(node.body, False)
-        else:
-            for child in ast.iter_child_nodes(node):
-                expression(child, present)
-
-    def block(statements, present):
-        for statement in statements:
-            if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)):
-                block(statement.body, False)
-                for value in statement.decorator_list + statement.args.defaults:
-                    expression(value, present)
-            elif isinstance(statement, ast.ClassDef):
-                block(statement.body, present)
-            elif isinstance(statement, ast.If):
-                expression(statement.test, present)
-                yes = block(statement.body, present or _condition_requires_procfs(statement.test, True))
-                no = block(statement.orelse, present or _condition_requires_procfs(statement.test, False))
-                continuing = [fact for fact in (yes, no) if fact is not None]
-                if not continuing:
-                    return None
-                present = all(continuing)
-            elif isinstance(statement, (ast.Return, ast.Raise, ast.Break, ast.Continue)):
-                for child in ast.iter_child_nodes(statement):
-                    expression(child, present)
-                return None
-            elif isinstance(statement, (ast.For, ast.AsyncFor, ast.While)):
-                condition = statement.test if isinstance(statement, ast.While) else statement.iter
-                expression(condition, present)
-                body_present = present or (isinstance(statement, ast.While) and _condition_requires_procfs(condition, True))
-                block(statement.body, body_present)
-                block(statement.orelse, present)
-            elif isinstance(statement, (ast.Try, getattr(ast, "TryStar", ast.Try))):
-                block(statement.body, present)
-                block(statement.orelse, present)
-                block(statement.finalbody, present)
-                for handler in statement.handlers:
-                    block(handler.body, present)
-            elif isinstance(statement, (ast.With, ast.AsyncWith)):
-                for item in statement.items:
-                    expression(item.context_expr, present)
-                block(statement.body, present)
-            else:
-                expression(statement, present)
-        return present
-
-    block(tree.body, False)
-    return guarded
+def _enclosing_scope(tree: ast.AST, node: ast.AST) -> ast.AST:
+    """Smallest function/module scope containing ``node``."""
+    best = tree
+    for candidate in ast.walk(tree):
+        if not isinstance(candidate, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            continue
+        end = getattr(candidate, "end_lineno", None) or candidate.lineno
+        if candidate.lineno <= node.lineno <= end:
+            if best is tree or candidate.lineno > best.lineno:
+                best = candidate
+    return best
 
 
 def _app_python_files() -> list[Path]:
@@ -216,10 +139,12 @@ def _collect_sites() -> tuple[list[str], list[str]]:
         sites = _scan_sites(tree)
         if not sites:
             continue
-        guards = _guarded_scan_ids(tree)
+        guards = _guard_lines(tree)
         for site in sites:
+            scope = _enclosing_scope(tree, site)
+            start = getattr(scope, "lineno", 0)
             label = f"{path.relative_to(REPO_ROOT)}:{site.lineno}"
-            if id(site) in guards:
+            if any(start <= g < site.lineno for g in guards):
                 guarded.append(label)
             else:
                 unguarded.append(label)
@@ -229,7 +154,7 @@ def _collect_sites() -> tuple[list[str], list[str]]:
 def test_every_procfs_scan_is_guarded_by_an_existence_check():
     _guarded, unguarded = _collect_sites()
     assert not unguarded, (
-        "procfs pid scans without an effective conditional existence guard — "
+        "procfs pid scans with no existence check in the enclosing function — "
         "these raise FileNotFoundError on macOS and Windows: "
         + ", ".join(unguarded)
     )
@@ -251,49 +176,3 @@ def test_the_guard_detector_still_sees_the_known_scans():
         f"root was renamed. Found: {sorted(found)}"
     )
     assert len(found) >= 3, f"expected at least 3 procfs scans, found {sorted(found)}"
-
-
-@pytest.mark.parametrize("body", [
-    'Path("/proc").exists(); os.listdir("/proc")',
-    'if Path("/proc").exists():\n    pass\nos.listdir("/proc")',
-    'if not Path("/proc").exists():\n    os.listdir("/proc")',
-    'if ready or Path("/proc").exists():\n    os.listdir("/proc")',
-    'if ready:\n    Path("/proc").exists()\nos.listdir("/proc")',
-    'os.listdir("/proc"); Path("/proc").exists()',
-    'if Path("/tmp").exists():\n    os.listdir("/proc")',
-    'def other():\n    Path("/proc").exists()\nos.listdir("/proc")',
-    'if not has_procfs():\n    pass\nos.listdir("/proc")',
-    'for x in []:\n    if not has_procfs():\n        return\nos.listdir("/proc")',
-    'if has_procfs():\n    return\nelse:\n    os.listdir("/proc")',
-    '[pid for pid in os.listdir("/proc") if has_procfs()]',
-], ids=["ignored", "expired-branch", "inverted", "unsafe-or", "sibling-branch",
-        "too-late", "wrong-path", "other-function", "no-early-exit", "loop-may-not-run",
-        "wrong-else", "late-comprehension-filter"])
-def test_guard_detector_rejects_ineffective_checks(body):
-    tree = ast.parse("def scan():\n" + "\n".join("    " + line for line in body.splitlines()))
-    sites = _scan_sites(tree)
-    assert sites, "negative control must contain a recognized enumeration"
-    assert any(id(site) not in _guarded_scan_ids(tree) for site in sites)
-
-
-@pytest.mark.parametrize("body", [
-    'if Path("/proc").exists():\n    os.listdir("/proc")',
-    'if not Path("/proc").exists():\n    return\nos.listdir("/proc")',
-    'if not has_procfs():\n    raise RuntimeError()\nos.listdir(PROC_ROOT)',
-    'if has_procfs():\n    pass\nelse:\n    return\nPROC_ROOT.iterdir()',
-    'if ready and has_procfs():\n    os.scandir("/proc")',
-    'if not has_procfs() or disabled:\n    return\nPROC_ROOT.glob("*")',
-    'if sys.platform == "linux" and has_procfs():\n    PROC_ROOT.iterdir()',
-    'for item in items:\n    if not has_procfs():\n        continue\n    os.listdir("/proc")',
-    'if os.path.isdir("/proc"):\n    os.listdir("/proc")',
-    'if has_procfs() and list(PROC_ROOT.iterdir()):\n    pass',
-    'if not has_procfs() or list(PROC_ROOT.iterdir()):\n    pass',
-    '[pid for item in items if has_procfs() for pid in os.listdir("/proc")]',
-], ids=["positive-branch", "early-return", "early-raise", "else-exits", "and",
-        "negative-or", "platform-and-procfs", "loop-continue", "isdir",
-        "short-circuit-and", "short-circuit-or", "guarded-comprehension"])
-def test_guard_detector_preserves_effective_checks(body):
-    tree = ast.parse("def scan():\n" + "\n".join("    " + line for line in body.splitlines()))
-    sites = _scan_sites(tree)
-    assert sites, "positive control must contain a recognized enumeration"
-    assert all(id(site) in _guarded_scan_ids(tree) for site in sites)