test(cdp): guard target teardown resource lifetimes

moli-cdp-smoke/README.md

@@ -164,6 +164,20 @@ and restricted-float conversion contracts. It was calibrated on 2026-09-07
 against Debian `/usr/bin/chromium` 145.0.7632.116 and runs independently of
 IndexedDB startup coverage.
 
+The default `target-lifecycle` process group locks down Moli's resource lifetime,
+not a Chromium-specific FD count. In one server it closes 800 default-context
+targets with `Target.closeTarget`, 800 with `Page.close`, 128 after detaching,
+and 64 by disposing explicit contexts. Foreground/background creation alternates.
+It waits for each exact `Target.targetDestroyed`, records Linux `/proc/<pid>/fd`
+and thread counts every batch, and checks a fixed post-warmup resource budget.
+Each phase must still load a real HTTP document and preserve a live peer Page.
+This catches closed renderer wakers retaining Tokio I/O drivers without changing
+the test to use a fresh context for each default-context Page. Batch progress,
+resource samples, and the usual server logs are retained on failure. With an
+external endpoint or on non-Linux systems, protocol churn/navigation still run;
+the artifact explicitly reports that FD sampling was unavailable. CI uses the
+managed Linux server, so the resource assertions are mandatory there.
+
 Covered well:
 
 - The default raw `debugger-breakpoints`, `runtime-exception`, and

moli-cdp-smoke/moli_cdp_smoke/groups/target_lifecycle.py

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+from typing import Any
+
+from ..assertions import SmokeError, assert_equal, record
+from ..raw_cdp import RawCdpClient, connect_raw_cdp
+from ..serve import MoliServe
+
+
+def process_resources(pid: int) -> dict[str, int] | None:
+    if not sys.platform.startswith("linux"):
+        return None
+    # Inspect only the managed server, not this client or other smoke workers.
+    # A missing/inaccessible managed process is a failure, not a skipped check.
+    descriptors: dict[int, str] = {}
+    for path in Path(f"/proc/{pid}/fd").iterdir():
+        try:
+            descriptors[int(path.name)] = str(path.readlink())
+        except FileNotFoundError:
+            # A transient socket can close between directory listing/readlink.
+            continue
+    return {
+        "fds": len(descriptors),
+        "eventpoll": sum(value == "anon_inode:[eventpoll]" for value in descriptors.values()),
+        "eventfd": sum(value == "anon_inode:[eventfd]" for value in descriptors.values()),
+        "threads": len(list(Path(f"/proc/{pid}/task").iterdir())),
+        "maxFd": max(descriptors, default=-1),
+    }
+
+
+def assert_resources_bounded(baseline: dict[str, int], current: dict[str, int]) -> None:
+    # Permit a fixed amount of in-flight teardown/transport bookkeeping, never
+    # a per-iteration allowance. The old +2 FD/page leak fails at the first batch.
+    for key, allowance in (("fds", 8), ("eventpoll", 2), ("eventfd", 2), ("threads", 2)):
+        if current[key] > baseline[key] + allowance:
+            raise SmokeError(f"target teardown leaked {key}: baseline={baseline}, current={current}")
+
+
+class LifecycleProbe:
+    def __init__(self, client: RawCdpClient) -> None:
+        self.client = client
+        self.destroyed: set[str] = set()
+        self.loaded: set[str] = set()
+
+    def observe(self, message: dict[str, Any]) -> None:
+        if message.get("method") == "Target.targetDestroyed":
+            self.destroyed.add(message["params"]["targetId"])
+        elif message.get("method") == "Page.loadEventFired":
+            self.loaded.add(message["sessionId"])
+
+    async def call(self, method: str, params: dict[str, Any] | None = None,
+                   session: str | None = None) -> dict[str, Any]:
+        response, seen = await self.client.recv_until_id(
+            await self.client.send(method, params, session_id=session), timeout=10,
+        )
+        for message in seen:
+            self.observe(message)
+        return response["result"]
+
+    async def wait_for(self, identities: set[str], identity: str, label: str) -> None:
+        async def receive() -> None:
+            while identity not in identities:
+                self.observe(await self.client.recv())
+            identities.remove(identity)
+        try:
+            await asyncio.wait_for(receive(), timeout=10)
+        except TimeoutError as error:
+            raise SmokeError(f"missing {label} for exact identity {identity}") from error
+
+    async def attach(self, target: str) -> str:
+        return (await self.call("Target.attachToTarget", {
+            "targetId": target, "flatten": True,
+        }))["sessionId"]
+
+    async def close(self, target: str) -> None:
+        result = await self.call("Target.closeTarget", {"targetId": target})
+        assert_equal(result.get("success"), True, "Target.closeTarget accepted")
+        await self.wait_for(self.destroyed, target, "Target.targetDestroyed")
+
+    async def cycle(self, mode: str, index: int) -> None:
+        params: dict[str, Any] = {"url": "about:blank", "background": bool(index % 2)}
+        if mode == "context":
+            params.update(await self.call("Target.createBrowserContext"))
+        target = (await self.call("Target.createTarget", params))["targetId"]
+        if mode == "context":
+            await self.call("Target.disposeBrowserContext", {
+                "browserContextId": params["browserContextId"],
+            })
+            await self.wait_for(self.destroyed, target, "disposed context target")
+        elif mode == "page":
+            await self.call("Page.close", session=await self.attach(target))
+            await self.wait_for(self.destroyed, target, "Page.close target destruction")
+        else:
+            if mode == "detach":
+                await self.call("Target.detachFromTarget", {"sessionId": await self.attach(target)})
+            await self.close(target)
+
+    async def navigate(self, url: str) -> None:
+        target = (await self.call("Target.createTarget", {"url": "about:blank"}))["targetId"]
+        session = await self.attach(target)
+        await self.call("Page.enable", session=session)
+        response = await self.call("Page.navigate", {"url": url}, session)
+        if response.get("errorText"):
+            raise SmokeError(f"navigation after target churn failed: {response}")
+        await self.wait_for(self.loaded, session, "post-churn Page.loadEventFired")
+        value = await self.call("Runtime.evaluate", {
+            "expression": "document.querySelector('main')?.textContent", "returnByValue": True,
+        }, session)
+        assert_equal(value.get("result", {}).get("value"), "plain ok", "real HTTP document after churn")
+        await self.close(target)
+
+
+async def run_target_lifecycle_group(
+    endpoint: str, fixture: str, results: list[dict[str, Any]], serve: MoliServe | None,
+) -> None:
+    pid = serve.process.pid if serve is not None else None
+    client = await connect_raw_cdp(endpoint)
+    probe = LifecycleProbe(client)
+    try:
+        await probe.call("Target.setDiscoverTargets", {"discover": True})
+        for target in (await probe.call("Target.getTargets"))["targetInfos"]:
+            if target["type"] == "page":
+                await probe.close(target["targetId"])
+        peer = (await probe.call("Target.createTarget", {"url": "about:blank"}))["targetId"]
+        peer_session = await probe.attach(peer)
+        await probe.call("Runtime.evaluate", {"expression": "globalThis.lifecycleSentinel = 42"}, peer_session)
+
+        modes = (("target", 800), ("page", 800), ("detach", 128), ("context", 64))
+        # Warm all closure routes and the shared network machinery once, then
+        # hold the same default BrowserContext and a live peer for every batch.
+        for mode, _ in modes:
+            for index in range(4):
+                await probe.cycle(mode, index)
+        await probe.navigate(f"{fixture}/plain?lifecycle-warmup")
+        baseline = process_resources(pid) if pid is not None else None
+        record(results, "target_lifecycle_baseline", {
+            "pid": pid, "resources": baseline,
+            "fdSampling": ("external-endpoint-without-owned-pid" if pid is None else
+                           "linux-proc" if baseline is not None else "unavailable-on-this-platform"),
+        })
+        for mode, count in modes:
+            for index in range(count):
+                await probe.cycle(mode, index)
+                if (index + 1) % 100 == 0 or index + 1 == count:
+                    current = process_resources(pid) if pid is not None else None
+                    # Persist the observation before asserting, so a failed
+                    # run retains the resource slope and exact failing batch.
+                    record(results, "target_lifecycle_sample", {
+                        "mode": mode, "closed": index + 1, "resources": current,
+                    })
+                    print(f"[moli-cdp-smoke] target-lifecycle {mode} {index + 1}/{count} {current}",
+                          file=sys.stderr, flush=True)
+                    if baseline is not None and current is not None:
+                        assert_resources_bounded(baseline, current)
+
+            await probe.navigate(f"{fixture}/plain?after-{mode}-{count}")
+            sentinel = await probe.call("Runtime.evaluate", {
+                "expression": "globalThis.lifecycleSentinel", "returnByValue": True,
+            }, peer_session)
+            assert_equal(sentinel.get("result", {}).get("value"), 42, "peer renderer survives teardown")
+            targets = (await probe.call("Target.getTargets"))["targetInfos"]
+            assert_equal([item["targetId"] for item in targets if item["type"] == "page"],
+                         [peer], "no closed Page remains registered")
+            record(results, f"target_lifecycle_{mode}", {"closed": count, "navigation": "ok", "peer": "alive"})
+        await probe.close(peer)
+    finally:
+        await client.websocket.close()

moli-cdp-smoke/moli_cdp_smoke/runner.py

@@ -62,6 +62,7 @@
 from .groups.stagehand import run_stagehand_group
 from .groups.svg_rect import run_svg_rect_group
 from .groups.target_semantics import run_target_semantics_group
+from .groups.target_lifecycle import run_target_lifecycle_group
 from .groups.tracing import run_raw_tracing_group, run_tracing_group
 from .groups.url_policy import run_url_policy_group
 from .groups.webgl_viewport import run_webgl_viewport_group
@@ -87,14 +88,15 @@
 ExternalGroupRunner = Callable[[str, str, list[dict[str, Any]]], Awaitable[None]]
 PageGroupRunner = Callable[[SmokeState], Awaitable[None]]
 BrowserGroupRunner = Callable[[Any, str, list[dict[str, Any]]], Awaitable[None]]
+ProcessGroupRunner = Callable[[str, str, list[dict[str, Any]], MoliServe | None], Awaitable[None]]
 
 
 @dataclass(frozen=True)
 class SmokeGroup:
     name: str
     description: str
     phase: str
-    runner: RawGroupRunner | ExternalGroupRunner | PageGroupRunner | BrowserGroupRunner
+    runner: RawGroupRunner | ExternalGroupRunner | PageGroupRunner | BrowserGroupRunner | ProcessGroupRunner
 
 
 async def _await_group(group: SmokeGroup, awaitable: Awaitable[None]) -> None:
@@ -417,8 +419,13 @@ async def _await_group(group: SmokeGroup, awaitable: Awaitable[None]) -> None:
 )
 
 
+PROCESS_GROUPS: tuple[SmokeGroup, ...] = (
+    SmokeGroup("target-lifecycle", "Managed-process target churn, FD bounds, and post-close navigation.",
+               "process", run_target_lifecycle_group),
+)
+
 DEFAULT_GROUPS: tuple[SmokeGroup, ...] = (
-    RAW_GROUPS + PAGE_GROUPS + BROWSER_GROUPS + MANAGED_EXTERNAL_GROUPS
+    RAW_GROUPS + PAGE_GROUPS + BROWSER_GROUPS + MANAGED_EXTERNAL_GROUPS + PROCESS_GROUPS
 )
 ALL_GROUPS: tuple[SmokeGroup, ...] = DEFAULT_GROUPS + OPTIONAL_EXTERNAL_GROUPS
 DEFAULT_GROUP_NAMES: tuple[str, ...] = tuple(group.name for group in DEFAULT_GROUPS)
@@ -430,6 +437,10 @@ async def _await_group(group: SmokeGroup, awaitable: Awaitable[None]) -> None:
 class SmokeSelection:
     groups: tuple[SmokeGroup, ...]
 
+    @property
+    def process_groups(self) -> tuple[SmokeGroup, ...]:
+        return tuple(group for group in self.groups if group.phase == "process")
+
     @property
     def raw_groups(self) -> tuple[SmokeGroup, ...]:
         return tuple(group for group in self.groups if group.phase == "raw")
@@ -671,6 +682,11 @@ async def async_main(argv: list[str] | None = None) -> int:
             raise RuntimeError("CDP endpoint was not initialized")
         if serve is None:
             await wait_for_cdp_server(endpoint, serve)
+        for current_group in selection.process_groups:
+            await _await_group(
+                current_group,
+                current_group.runner(endpoint, fixture.url, results, serve),  # type: ignore[misc]
+            )
         for current_group in selection.raw_groups:
             await _await_group(
                 current_group,

moli-cdp-smoke/tests/test_group_selection.py

@@ -34,6 +34,7 @@ def test_default_runs_every_repository_managed_group(self) -> None:
         self.assertIn("media-error", DEFAULT_GROUP_NAMES)
         self.assertIn("webgl-viewport", DEFAULT_GROUP_NAMES)
         self.assertIn("svg-rect", DEFAULT_GROUP_NAMES)
+        self.assertIn("target-lifecycle", DEFAULT_GROUP_NAMES)
         self.assertIn("multi-page", DEFAULT_GROUP_NAMES)
         self.assertIn("puppeteer", DEFAULT_GROUP_NAMES)
         self.assertEqual(
@@ -50,6 +51,12 @@ def test_only_external_environment_groups_remain_opt_in(self) -> None:
         )
         self.assertTrue(set(optional_names).isdisjoint(DEFAULT_GROUP_NAMES))
 
+    def test_target_lifecycle_uses_managed_process_without_playwright_context(self) -> None:
+        selection = resolve_group_selection(["target-lifecycle"])
+        self.assertEqual(tuple(group.name for group in selection.process_groups), ("target-lifecycle",))
+        self.assertFalse(selection.needs_playwright)
+        self.assertFalse(selection.raw_groups)
+
     def test_registry_and_listing_mark_exact_default_set(self) -> None:
         self.assertEqual(len(ALL_GROUPS), len(GROUPS_BY_NAME))
         listed_defaults = tuple(

moli-cdp-smoke/tests/test_target_lifecycle.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from moli_cdp_smoke.assertions import SmokeError
+from moli_cdp_smoke.groups.target_lifecycle import (
+    LifecycleProbe, assert_resources_bounded, process_resources,
+)
+
+
+class TargetLifecycleTests(unittest.TestCase):
+    def test_resource_budget_is_fixed_not_proportional_to_iterations(self) -> None:
+        baseline = {"fds": 24, "eventpoll": 7, "eventfd": 4, "threads": 20}
+        assert_resources_bounded(baseline, baseline)
+        assert_resources_bounded(baseline, dict(baseline, fds=28, eventpoll=8, eventfd=5))
+        for key, value in (("fds", 224), ("eventpoll", 10), ("eventfd", 7), ("threads", 23)):
+            with self.subTest(key=key), self.assertRaises(SmokeError):
+                assert_resources_bounded(baseline, dict(baseline, **{key: value}))
+
+    def test_event_identity_is_target_and_session_specific(self) -> None:
+        probe = LifecycleProbe(None)  # type: ignore[arg-type]
+        probe.observe({"method": "Target.targetDestroyed", "params": {"targetId": "old"}})
+        probe.observe({"method": "Page.loadEventFired", "sessionId": "session-A"})
+        probe.observe({"method": "Target.targetCreated", "params": {"targetInfo": {"targetId": "new"}}})
+        self.assertEqual(probe.destroyed, {"old"})
+        self.assertEqual(probe.loaded, {"session-A"})
+        self.assertNotIn("session-B", probe.loaded)
+
+    def test_non_linux_fd_sampling_is_explicitly_unavailable(self) -> None:
+        with patch("moli_cdp_smoke.groups.target_lifecycle.sys.platform", "darwin"):
+            self.assertIsNone(process_resources(1234))
+
+    def test_missing_managed_process_is_not_silently_skipped(self) -> None:
+        with patch("moli_cdp_smoke.groups.target_lifecycle.sys.platform", "linux"), \
+             patch("moli_cdp_smoke.groups.target_lifecycle.Path.iterdir", side_effect=FileNotFoundError):
+            with self.assertRaises(FileNotFoundError):
+                process_resources(1234)