fix(browser): bound forced-close teardown to one deadline

scripts/ci/run_mv3_compatibility.py

@@ -621,6 +621,76 @@ def _wait_for_linux_process_identity_set_exit(
         time.sleep(min(0.05, remaining_seconds))
 
 
+def _wait_for_linux_process_teardown(
+    root_process_id: int,
+    root_start_time_ticks: int,
+    process_identities: tuple[tuple[int, int], ...],
+    *,
+    timeout_seconds: float = PROCESS_EXIT_TIMEOUT_SECONDS,
+) -> tuple[bool, bool]:
+    """Observe root and sampled-set termination under one bounded shared deadline."""
+
+    if (
+        isinstance(root_process_id, bool)
+        or not isinstance(root_process_id, int)
+        or root_process_id <= 0
+    ):
+        raise ValueError("invalid Linux root process identifier")
+    if (
+        isinstance(root_start_time_ticks, bool)
+        or not isinstance(root_start_time_ticks, int)
+        or root_start_time_ticks <= 0
+    ):
+        raise ValueError("invalid Linux root process start time")
+    if not process_identities or len(process_identities) > MAX_BROWSER_PROCESS_TREE_SIZE:
+        raise ValueError("invalid Linux process identity-set size")
+
+    expected: dict[int, tuple[int, int]] = {}
+    for identity in process_identities:
+        if not isinstance(identity, tuple) or len(identity) != 2:
+            raise ValueError("invalid Linux process identity")
+        process_id, start_time_ticks = identity
+        if isinstance(process_id, bool) or not isinstance(process_id, int) or process_id <= 0:
+            raise ValueError("invalid Linux process identifier")
+        if (
+            isinstance(start_time_ticks, bool)
+            or not isinstance(start_time_ticks, int)
+            or start_time_ticks <= 0
+        ):
+            raise ValueError("invalid Linux process start time")
+        if process_id in expected:
+            raise ValueError("Linux process identity-set PIDs must be unique")
+        expected[process_id] = identity
+
+    root_identity = (root_process_id, root_start_time_ticks)
+    if expected.get(root_process_id) != root_identity:
+        raise ValueError("Linux root process identity must belong to the sampled process set")
+    if (
+        isinstance(timeout_seconds, bool)
+        or not isinstance(timeout_seconds, (int, float))
+        or timeout_seconds < 0
+        or not math.isfinite(timeout_seconds)
+    ):
+        raise ValueError("invalid Linux process teardown timeout")
+
+    deadline = time.monotonic() + float(timeout_seconds)
+    while True:
+        live_process_ids: set[int] = set()
+        for process_id, expected_identity in expected.items():
+            current_identity = _read_linux_proc_stat_process_identity(process_id)
+            if current_identity == expected_identity:
+                live_process_ids.add(process_id)
+
+        root_terminated = root_process_id not in live_process_ids
+        process_set_terminated = not live_process_ids
+        if process_set_terminated:
+            return root_terminated, True
+        remaining_seconds = deadline - time.monotonic()
+        if remaining_seconds <= 0:
+            return root_terminated, False
+        time.sleep(min(0.05, remaining_seconds))
+
+
 def _sample_linux_process_rss_bytes(process_id: int) -> int:
     """Read one attributed Linux process RSS through a bounded ``/proc`` status file."""
 
@@ -1779,15 +1849,22 @@ def _run_agent_task_forced_close_browser_pass(
 
     if browser_process_id is None or browser_process_start_time_ticks is None:
         raise RuntimeError("Agent Task forced-close browser process identity was not captured")
-    browser_process_terminated = _wait_for_linux_process_identity_exit(
-        browser_process_id,
-        browser_process_start_time_ticks,
+    full_process_set_captured = chromium_process_identities is not None
+    teardown_identities = (
+        chromium_process_identities
+        if chromium_process_identities is not None
+        else ((browser_process_id, browser_process_start_time_ticks),)
     )
-    chromium_process_set_terminated: bool | None = None
-    if chromium_process_identities is not None:
-        chromium_process_set_terminated = _wait_for_linux_process_identity_set_exit(
-            chromium_process_identities
+    browser_process_terminated, observed_process_set_terminated = (
+        _wait_for_linux_process_teardown(
+            browser_process_id,
+            browser_process_start_time_ticks,
+            teardown_identities,
         )
+    )
+    chromium_process_set_terminated = (
+        observed_process_set_terminated if full_process_set_captured else None
+    )
     if (
         browser_failure_type is not None
         or session_cleanup_failure_type is not None

tests/test_agent_task_forced_close_process_termination_contract.py

@@ -27,8 +27,7 @@ def test_forced_close_browser_pass_binds_and_waits_for_process_identities(self)
             "_snapshot_linux_process_evidence",
             "_read_linux_process_identity_set",
             "_terminate_owned_process_bounded",
-            "_wait_for_linux_process_identity_exit",
-            "_wait_for_linux_process_identity_set_exit",
+            "_wait_for_linux_process_teardown",
             '"driver_process_terminated"',
             '"driver_kill_fallback_used"',
             '"browser_process_terminated"',
@@ -97,12 +96,10 @@ def test_forced_close_browser_failure_is_returned_after_teardown_waits(self) ->
                 self.assertIn(expected, browser_pass)
 
         shutdown = browser_pass.index("_terminate_owned_process_bounded(driver)")
-        root_wait = browser_pass.index("_wait_for_linux_process_identity_exit(")
-        set_wait = browser_pass.index("_wait_for_linux_process_identity_set_exit(")
-        failure_return = browser_pass.index("browser_failure_type is not None", set_wait)
-        self.assertLess(shutdown, root_wait)
-        self.assertLess(root_wait, failure_return)
-        self.assertLess(set_wait, failure_return)
+        teardown_wait = browser_pass.index("_wait_for_linux_process_teardown(")
+        failure_return = browser_pass.index("browser_failure_type is not None", teardown_wait)
+        self.assertLess(shutdown, teardown_wait)
+        self.assertLess(teardown_wait, failure_return)
 
     def test_forced_close_driver_shutdown_timeout_is_bounded_and_typed(self) -> None:
         """A wedged ChromeDriver after SIGKILL must become failure evidence, not escape."""

tests/test_agent_task_forced_close_shared_teardown_deadline_contract.py

@@ -0,0 +1,90 @@
+"""Contract for one total post-shutdown teardown deadline in the forced-close lane."""
+
+from __future__ import annotations
+
+import pathlib
+import runpy
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py"
+
+
+class AgentTaskForcedCloseSharedTeardownDeadlineContractTests(unittest.TestCase):
+    """Prevent root and process-set teardown polling from multiplying the budget."""
+
+    def test_runner_exposes_one_combined_teardown_waiter(self) -> None:
+        """Root and sampled-set evidence must be observed under one timeout authority."""
+
+        namespace = runpy.run_path(
+            str(RUNNER), run_name="forced_close_shared_teardown_deadline"
+        )
+        self.assertIn("_wait_for_linux_process_teardown", namespace)
+
+    def test_combined_waiter_preserves_partial_evidence_at_one_deadline(self) -> None:
+        """A root may exit while a descendant remains live when the one deadline expires."""
+
+        namespace = runpy.run_path(
+            str(RUNNER), run_name="forced_close_shared_teardown_behavior"
+        )
+        waiter = namespace["_wait_for_linux_process_teardown"]
+
+        class FakeTime:
+            def __init__(self) -> None:
+                self.now = 0.0
+
+            def monotonic(self) -> float:
+                return self.now
+
+            def sleep(self, seconds: float) -> None:
+                self.now += seconds
+
+        fake_time = FakeTime()
+        root_identity = (101, 1_001)
+        child_identity = (202, 2_002)
+
+        def fake_read(process_id: int) -> tuple[int, int] | None:
+            if process_id == root_identity[0]:
+                return None if fake_time.now >= 0.05 else root_identity
+            if process_id == child_identity[0]:
+                return child_identity
+            raise AssertionError(f"unexpected process id: {process_id}")
+
+        waiter.__globals__["time"] = fake_time
+        waiter.__globals__["_read_linux_proc_stat_process_identity"] = fake_read
+
+        root_terminated, process_set_terminated = waiter(
+            root_identity[0],
+            root_identity[1],
+            (root_identity, child_identity),
+            timeout_seconds=0.10,
+        )
+        self.assertIs(root_terminated, True)
+        self.assertIs(process_set_terminated, False)
+        self.assertLessEqual(fake_time.now, 0.1000001)
+
+    def test_combined_waiter_requires_root_identity_in_the_sampled_set(self) -> None:
+        """A separate root identity may not be paired with an unrelated process set."""
+
+        namespace = runpy.run_path(
+            str(RUNNER), run_name="forced_close_shared_teardown_identity"
+        )
+        waiter = namespace["_wait_for_linux_process_teardown"]
+        with self.assertRaises(ValueError):
+            waiter(101, 1_001, ((202, 2_002),), timeout_seconds=0)
+
+    def test_forced_close_browser_pass_uses_only_the_combined_waiter(self) -> None:
+        """The forced-close pass must not run independent root and set timeout windows."""
+
+        runner = RUNNER.read_text(encoding="utf-8")
+        start = runner.index("def _run_agent_task_forced_close_browser_pass(")
+        end = runner.index("\ndef _run_agent_task_forced_close_trial(", start)
+        browser_pass = runner[start:end]
+
+        self.assertIn("_wait_for_linux_process_teardown(", browser_pass)
+        self.assertNotIn("_wait_for_linux_process_identity_exit(", browser_pass)
+        self.assertNotIn("_wait_for_linux_process_identity_set_exit(", browser_pass)
+
+
+if __name__ == "__main__":
+    unittest.main()