diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py old mode 100755 new mode 100644 index ce4b0b098..8b810caf4 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -1274,20 +1274,27 @@ def _run_agent_task_browser_pass( 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 + ) if browser_failure_type is not None: - return { + failure_evidence: dict[str, Any] = { "failure_type": browser_failure_type, "browser_process_terminated": browser_process_terminated, } + if chromium_process_set_terminated is not None: + failure_evidence["chromium_process_set_terminated"] = ( + chromium_process_set_terminated + ) + return failure_evidence if result is None: raise RuntimeError("Agent Task browser pass returned no result after shutdown") - if chromium_process_identities is None: + if chromium_process_set_terminated is None: raise RuntimeError("Agent Task Chromium process identities were not captured") if chromium_process_pre_shutdown_exit_count is None: raise RuntimeError("Agent Task Chromium pre-shutdown exit count was not captured") - chromium_process_set_terminated = _wait_for_linux_process_identity_set_exit( - chromium_process_identities - ) if not browser_process_terminated: raise RuntimeError("Agent Task browser process did not terminate") if not chromium_process_set_terminated: @@ -1350,7 +1357,7 @@ def _run_agent_task_trial( browser_process_terminated = result.get("browser_process_terminated") if not isinstance(browser_process_terminated, bool): raise RuntimeError("Agent Task browser pass returned invalid teardown evidence") - return { + failure_evidence: dict[str, Any] = { "trial_number": trial_number, "passed": False, "failure_type": returned_failure_type, @@ -1358,6 +1365,16 @@ def _run_agent_task_trial( "profile_cleaned": True, "duration_ms": duration_ms, } + if "chromium_process_set_terminated" in result: + chromium_process_set_terminated = result["chromium_process_set_terminated"] + if not isinstance(chromium_process_set_terminated, bool): + raise RuntimeError( + "Agent Task browser pass returned invalid process-set teardown evidence" + ) + failure_evidence["chromium_process_set_terminated"] = ( + chromium_process_set_terminated + ) + return failure_evidence return { "trial_number": trial_number, @@ -1455,6 +1472,10 @@ def _run_agent_task_forced_close_browser_pass( driver_port = _free_loopback_port() session_id: str | None = None + browser_process_id: int | None = None + browser_process_start_time_ticks: int | None = None + chromium_process_identities: tuple[tuple[int, int], ...] | None = None + result: dict[str, Any] | None = None driver = subprocess.Popen( [str(chromedriver_bin), f"--port={driver_port}", "--allowed-ips=127.0.0.1"], stdout=subprocess.DEVNULL, @@ -1499,11 +1520,22 @@ def _run_agent_task_forced_close_browser_pass( raise RuntimeError("ChromeDriver forced-close capabilities are malformed") session_id = _path_token(raw_session_id, "session identifier") browser_version = capabilities.get("browserVersion") + browser_process_id = capabilities.get("goog:processID") if browser_version != PINNED_CHROME_VERSION: raise RuntimeError( f"unexpected forced-close Chrome version: expected {PINNED_CHROME_VERSION}, " f"got {browser_version!r}" ) + if ( + isinstance(browser_process_id, bool) + or not isinstance(browser_process_id, int) + or browser_process_id <= 0 + ): + raise RuntimeError("ChromeDriver did not return a valid forced-close browser process id") + browser_process_identity = _read_linux_proc_stat_process_identity(browser_process_id) + if browser_process_identity is None: + raise RuntimeError("Agent Task forced-close browser process identity disappeared") + browser_process_start_time_ticks = browser_process_identity[1] survivor_context = _json_request( driver_port, @@ -1549,6 +1581,21 @@ def _run_agent_task_forced_close_browser_pass( if loaded_url != fixture_url: raise RuntimeError("Agent Task forced-close probe did not load its fixture URL") + process_evidence = _snapshot_linux_process_evidence() + chromium_process_ids = _discover_linux_process_tree_ids( + browser_process_id, + process_evidence, + ) + chromium_process_identities, _pre_shutdown_exit_count = ( + _read_linux_process_identity_set( + chromium_process_ids, + required_root_identity=( + browser_process_id, + browser_process_start_time_ticks, + ), + ) + ) + forced_close_detected = _force_close_agent_task_context(driver_port, session_id) if not forced_close_detected: raise RuntimeError("Agent Task forced-close probe did not detect the close") @@ -1567,7 +1614,7 @@ def _run_agent_task_forced_close_browser_pass( if not isinstance(surviving_url, str): raise RuntimeError("Agent Task survivor context was not usable after forced close") - return { + result = { "browser_version": browser_version, "forced_close_detected": forced_close_detected, "session_survived": True, @@ -1588,6 +1635,27 @@ def _run_agent_task_forced_close_browser_pass( driver.kill() driver.wait(timeout=5) + 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") + if chromium_process_identities is None: + raise RuntimeError("Agent Task forced-close Chromium process identities were not captured") + browser_process_terminated = _wait_for_linux_process_identity_exit( + browser_process_id, + browser_process_start_time_ticks, + ) + chromium_process_set_terminated = _wait_for_linux_process_identity_set_exit( + chromium_process_identities + ) + if result is None: + raise RuntimeError("Agent Task forced-close browser pass returned no result after shutdown") + if not browser_process_terminated: + raise RuntimeError("Agent Task forced-close browser process did not terminate") + if not chromium_process_set_terminated: + raise RuntimeError("Agent Task forced-close Chromium process set did not terminate") + result["browser_process_terminated"] = True + result["chromium_process_set_terminated"] = True + return result + def _run_agent_task_forced_close_trial( chrome_bin: pathlib.Path, @@ -1644,6 +1712,8 @@ def _run_agent_task_forced_close_trial( "browser_version": result["browser_version"], "forced_close_detected": result["forced_close_detected"], "session_survived": result["session_survived"], + "browser_process_terminated": result["browser_process_terminated"], + "chromium_process_set_terminated": result["chromium_process_set_terminated"], "profile_cleaned": True, "duration_ms": duration_ms, } @@ -1849,6 +1919,8 @@ def main() -> int: forced_close_surfaces_complete = all( trial.get("forced_close_detected") is True and trial.get("session_survived") is True + and trial.get("browser_process_terminated") is True + and trial.get("chromium_process_set_terminated") is True and trial.get("profile_cleaned") is True for trial in forced_close_trials if trial.get("passed") is True diff --git a/tests/test_agent_task_failure_process_set_termination_contract.py b/tests/test_agent_task_failure_process_set_termination_contract.py new file mode 100644 index 000000000..6fe48ea9d --- /dev/null +++ b/tests/test_agent_task_failure_process_set_termination_contract.py @@ -0,0 +1,94 @@ +"""Contract for Chromium process-set termination evidence after Agent Task failure.""" + +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 AgentTaskFailureProcessSetTerminationContractTests(unittest.TestCase): + """Retain sampled descendant teardown evidence when controlled browser work fails.""" + + def _namespace(self, name: str) -> dict[str, object]: + return runpy.run_path(str(RUNNER), run_name=name) + + def test_browser_pass_retains_sampled_process_set_teardown_after_failure(self) -> None: + """Late failure must not discard identities already captured before shutdown.""" + + runner = RUNNER.read_text(encoding="utf-8") + start = runner.index("def _run_agent_task_browser_pass(") + end = runner.index("\ndef _run_agent_task_trial(", start) + browser_pass = runner[start:end] + for expected in ( + "if chromium_process_identities is not None:", + "chromium_process_set_terminated = _wait_for_linux_process_identity_set_exit(", + 'failure_evidence["chromium_process_set_terminated"]', + ): + with self.subTest(expected=expected): + self.assertIn(expected, browser_pass) + + def test_trial_preserves_failed_process_set_teardown_evidence(self) -> None: + """Profile cleanup must preserve both root and sampled-set termination outcomes.""" + + namespace = self._namespace("agent_task_failure_process_set_termination_trial") + run_trial = namespace["_run_agent_task_trial"] + + def fail_after_sampled_set_shutdown( + *_args: object, **_kwargs: object + ) -> dict[str, object]: + return { + "failure_type": "RuntimeError", + "browser_process_terminated": True, + "chromium_process_set_terminated": False, + } + + run_trial.__globals__["_run_agent_task_browser_pass"] = ( + fail_after_sampled_set_shutdown + ) + result = run_trial( + pathlib.Path("controlled-chrome"), + pathlib.Path("controlled-chromedriver"), + "http://127.0.0.1/controlled-fixture", + 13, + ) + + self.assertEqual(result["trial_number"], 13) + self.assertIs(result["passed"], False) + self.assertEqual(result["failure_type"], "RuntimeError") + self.assertIs(result["browser_process_terminated"], True) + self.assertIs(result["chromium_process_set_terminated"], False) + self.assertIs(result["profile_cleaned"], True) + + def test_failure_before_process_set_capture_does_not_invent_set_evidence(self) -> None: + """A failure without sampled identities must remain explicit rather than fabricated.""" + + namespace = self._namespace("agent_task_failure_before_process_set_capture") + run_trial = namespace["_run_agent_task_trial"] + + def fail_before_process_set_capture( + *_args: object, **_kwargs: object + ) -> dict[str, object]: + return { + "failure_type": "RuntimeError", + "browser_process_terminated": True, + } + + run_trial.__globals__["_run_agent_task_browser_pass"] = fail_before_process_set_capture + result = run_trial( + pathlib.Path("controlled-chrome"), + pathlib.Path("controlled-chromedriver"), + "http://127.0.0.1/controlled-fixture", + 14, + ) + + self.assertIs(result["passed"], False) + self.assertNotIn("chromium_process_set_terminated", result) + self.assertIs(result["profile_cleaned"], True) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_agent_task_forced_close_process_termination_contract.py b/tests/test_agent_task_forced_close_process_termination_contract.py new file mode 100644 index 000000000..319227b98 --- /dev/null +++ b/tests/test_agent_task_forced_close_process_termination_contract.py @@ -0,0 +1,84 @@ +"""Contract for post-shutdown process termination in the Agent Task 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 AgentTaskForcedCloseProcessTerminationContractTests(unittest.TestCase): + """Require interruption evidence to include bounded Chromium teardown proof.""" + + def test_forced_close_browser_pass_binds_and_waits_for_process_identities(self) -> None: + """The forced-close pass must prove its sampled browser process set terminates.""" + + 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] + for expected in ( + 'capabilities.get("goog:processID")', + "_read_linux_proc_stat_process_identity", + "_snapshot_linux_process_evidence", + "_read_linux_process_identity_set", + "_wait_for_linux_process_identity_exit", + "_wait_for_linux_process_identity_set_exit", + '"browser_process_terminated"', + '"chromium_process_set_terminated"', + ): + with self.subTest(expected=expected): + self.assertIn(expected, browser_pass) + + def test_forced_close_trial_preserves_false_teardown_evidence(self) -> None: + """A failed teardown proof must not be omitted or normalized into success.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="forced_close_process_termination_trial" + ) + trial = namespace["_run_agent_task_forced_close_trial"] + + def fake_browser_pass( + _chrome_bin: pathlib.Path, + _chromedriver_bin: pathlib.Path, + _fixture_url: str, + _profile_dir: str, + ) -> dict[str, object]: + return { + "browser_version": namespace["PINNED_CHROME_VERSION"], + "forced_close_detected": True, + "session_survived": True, + "browser_process_terminated": False, + "chromium_process_set_terminated": False, + } + + trial.__globals__["_run_agent_task_forced_close_browser_pass"] = fake_browser_pass + result = trial( + pathlib.Path("/unused/chrome"), + pathlib.Path("/unused/chromedriver"), + "http://127.0.0.1/fixture", + 1, + ) + self.assertIs(result["browser_process_terminated"], False) + self.assertIs(result["chromium_process_set_terminated"], False) + + def test_main_forced_close_gate_requires_process_termination(self) -> None: + """Compatibility success must reject a live forced-close browser identity.""" + + runner = RUNNER.read_text(encoding="utf-8") + start = runner.index("forced_close_surfaces_complete = all(") + end = runner.index("\n\n evidence = {", start) + gate = runner[start:end] + for expected in ( + 'trial.get("browser_process_terminated") is True', + 'trial.get("chromium_process_set_terminated") is True', + ): + with self.subTest(expected=expected): + self.assertIn(expected, gate) + + +if __name__ == "__main__": + unittest.main()