fix(cdp-smoke): observe browser exit independently of output EOF

moli-cdp-smoke/README.md

@@ -23,6 +23,13 @@ Use managed `--browser chromium` runs for the full oracle, including process
 resource sampling; sharing an external `--endpoint` does not provide the same
 process isolation or owned PID.
 
+The `Browser.close` lifecycle contract requires an acknowledgment followed by a
+natural, zero-status exit of the owned browser process before cleanup. It
+observes the parent's return code independently of stdout/stderr EOF: a
+descendant may retain an inherited output pipe after the browser has exited,
+which can leave `asyncio.subprocess.Process.wait()` pending. Failures identify
+the session case and lifecycle stage, including acknowledgment and process exit.
+
 Lifecycle artifacts retain total FDs for both engines. Chromium also records
 FDs for regular files inside its owned profile: delayed database initialization
 opens these independently of target churn. Its fixed FD budget covers the

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

@@ -5,6 +5,7 @@ import json
 from typing import Any
 
 from ..assertions import SmokeError, assert_equal, wait_until
+from ..process import wait_for_process_exit
 from ..raw_cdp import RawCdpClient, connect_raw_cdp, connect_raw_cdp_websocket
 from ..serve import browser_identity, start_browser, stop_browser, wait_for_browser_endpoint
 
@@ -132,6 +133,7 @@ async def browser_close_instance_lifecycle(endpoint: str, fixture: str) -> dict[
         # scenario. The supplied endpoint is only an oracle and a survivor.
         serving = await start_browser()
         clients: list[RawCdpClient] = []
+        stage = "startup"
         try:
             child_endpoint = await wait_for_browser_endpoint(serving)
             await browser_identity(child_endpoint, identity["engine"])
@@ -139,6 +141,7 @@ async def browser_close_instance_lifecycle(endpoint: str, fixture: str) -> dict[
             clients.append(root)
             rejected = None
             if scope in ("unknown", "null", "foreign"):
+                stage = "invalid session rejection"
                 selected, session = root, _OMITTED
                 if scope == "foreign":
                     attached = await _command(root, "Target.attachToBrowserTarget")
@@ -152,17 +155,20 @@ async def browser_close_instance_lifecycle(endpoint: str, fixture: str) -> dict[
                 await _command(root, "Browser.getVersion")
                 assert_equal(serving.process.returncode, None, "invalid command keeps instance alive")
             else:
+                stage = "session setup"
                 selected, session = await _route(root, child_endpoint, fixture, scope, clients)
+            stage = "close acknowledgment"
             response = await _close_request(selected, params=params, session=session)
             expected: dict[str, Any] = {"id": 1000, "result": {}}
             if session is not _OMITTED:
                 expected["sessionId"] = session
             assert_equal(response, expected, "close acknowledgment before EOF")
-            code = await asyncio.wait_for(serving.process.wait(), 10)
+            stage = "natural process exit"
+            code = await wait_for_process_exit(serving.process, timeout_seconds=10)
             assert_equal(code, 0, "natural process exit")
             observed.append({"case": name, "response": response, "exitCode": code, "rejected": rejected})
         except Exception as error:
-            raise SmokeError(f"{name}: {type(error).__name__}: {error}") from error
+            raise SmokeError(f"{name} ({stage}): {type(error).__name__}: {error}") from error
         finally:
             await asyncio.gather(
                 *(client.websocket.close() for client in clients), return_exceptions=True,

moli-cdp-smoke/moli_cdp_smoke/process.py

@@ -39,6 +39,19 @@ class CapturedProcessTimeout(TimeoutError):
         self.output_closed = output_closed
 
 
+async def wait_for_process_exit(
+    process: asyncio.subprocess.Process,
+    *,
+    timeout_seconds: float,
+) -> int:
+    # Process.wait() may also wait for stdout/stderr EOF. Descendants can keep
+    # those pipes open after the owned browser has exited and been reaped.
+    async with asyncio.timeout(timeout_seconds):
+        while process.returncode is None:
+            await asyncio.sleep(0.01)
+    return process.returncode
+
+
 def _consume_task_result(task: asyncio.Task[object]) -> None:
     try:
         task.exception()

moli-cdp-smoke/tests/test_external_process.py

@@ -9,7 +9,7 @@ import unittest
 
 from moli_cdp_smoke.assertions import SmokeError
 from moli_cdp_smoke.groups.external_process import run_external_json_process
-from moli_cdp_smoke.process import terminate_process_tree
+from moli_cdp_smoke.process import terminate_process_tree, wait_for_process_exit
 
 
 class ExternalProcessTests(unittest.IsolatedAsyncioTestCase):
@@ -83,5 +83,63 @@ class ExternalProcessTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(process.returncode, -signal.SIGKILL)
 
 
+class NaturalProcessExitTests(unittest.IsolatedAsyncioTestCase):
+    @unittest.skipUnless(os.name == "posix", "inherited-pipe regression is POSIX-specific")
+    async def test_parent_exit_does_not_wait_for_descendant_output_eof(self) -> None:
+        descendant = "import time; time.sleep(30)"
+        parent = (
+            "import subprocess, sys; "
+            f"subprocess.Popen([sys.executable, '-c', {descendant!r}]); "
+            "print('ready', flush=True); sys.stdin.read(1)"
+        )
+        process = await asyncio.create_subprocess_exec(
+            sys.executable, "-c", parent,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            start_new_session=True,
+        )
+        pipe_wait = None
+        try:
+            self.assertEqual(await asyncio.wait_for(process.stdout.readline(), 5), b"ready\n")
+            pipe_wait = asyncio.create_task(process.wait())
+            await asyncio.sleep(0)
+            process.stdin.write(b"x")
+            await process.stdin.drain()
+
+            code = await wait_for_process_exit(process, timeout_seconds=5)
+
+            self.assertEqual(code, 0)
+            self.assertFalse(pipe_wait.done(), "descendant still owns stdout/stderr")
+        finally:
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+            await asyncio.wait_for(pipe_wait if pipe_wait is not None else process.wait(), 5)
+
+    async def test_exit_status_is_preserved(self) -> None:
+        process = await asyncio.create_subprocess_exec(
+            sys.executable, "-c", "raise SystemExit(7)",
+        )
+
+        self.assertEqual(await wait_for_process_exit(process, timeout_seconds=5), 7)
+
+    async def test_deadline_does_not_terminate_a_live_process(self) -> None:
+        process = await asyncio.create_subprocess_exec(
+            sys.executable, "-c",
+            "import time; print('ready', flush=True); time.sleep(30)",
+            stdout=asyncio.subprocess.PIPE,
+        )
+        try:
+            self.assertEqual(await asyncio.wait_for(process.stdout.readline(), 5), b"ready\n")
+            with self.assertRaises(TimeoutError):
+                await wait_for_process_exit(process, timeout_seconds=0.02)
+            self.assertIsNone(process.returncode)
+        finally:
+            process.kill()
+            await process.communicate()
+
+
 if __name__ == "__main__":
     unittest.main()