diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index e5cc239af..73880874b 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -923,6 +923,7 @@ def _run_agent_task_browser_pass( session_id: str | None = None browser_process_id: int | None = None browser_process_start_time_ticks: int | None = None + browser_failure_type: str | None = None result: dict[str, Any] | None = None driver = subprocess.Popen( [str(chromedriver_bin), f"--port={driver_port}", "--allowed-ips=127.0.0.1"], @@ -1136,6 +1137,10 @@ def _run_agent_task_browser_pass( "task_duration_ms": task_duration_ms, "duration_ms": round(task_duration_ms), } + except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc: + browser_failure_type = type(exc).__name__ + if browser_process_id is None or browser_process_start_time_ticks is None: + raise finally: if session_id is not None: with contextlib.suppress(Exception): @@ -1152,14 +1157,20 @@ def _run_agent_task_browser_pass( driver.kill() driver.wait(timeout=5) - if result is None: - raise RuntimeError("Agent Task browser pass returned no result after shutdown") if browser_process_id is None or browser_process_start_time_ticks is None: raise RuntimeError("Agent Task browser process identity was not captured") - if not _wait_for_linux_process_identity_exit( + browser_process_terminated = _wait_for_linux_process_identity_exit( browser_process_id, browser_process_start_time_ticks, - ): + ) + if browser_failure_type is not None: + return { + "failure_type": browser_failure_type, + "browser_process_terminated": browser_process_terminated, + } + if result is None: + raise RuntimeError("Agent Task browser pass returned no result after shutdown") + if not browser_process_terminated: raise RuntimeError("Agent Task browser process did not terminate") result["browser_process_terminated"] = True return result @@ -1211,6 +1222,21 @@ def _run_agent_task_trial( } if result is None: raise RuntimeError("Agent Task browser pass returned no result") + returned_failure_type = result.get("failure_type") + if returned_failure_type is not None: + if not isinstance(returned_failure_type, str) or not returned_failure_type: + raise RuntimeError("Agent Task browser pass returned invalid failure evidence") + 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 { + "trial_number": trial_number, + "passed": False, + "failure_type": returned_failure_type, + "browser_process_terminated": browser_process_terminated, + "profile_cleaned": True, + "duration_ms": duration_ms, + } return { "trial_number": trial_number, @@ -1767,4 +1793,4 @@ def main() -> int: if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file diff --git a/tests/test_agent_task_failure_process_termination_contract.py b/tests/test_agent_task_failure_process_termination_contract.py new file mode 100644 index 000000000..7aae4d475 --- /dev/null +++ b/tests/test_agent_task_failure_process_termination_contract.py @@ -0,0 +1,89 @@ +"""Contract for browser-process 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 AgentTaskFailureProcessTerminationContractTests(unittest.TestCase): + """Require failed browser work to retain exact root-process teardown evidence.""" + + def _namespace(self, name: str) -> dict[str, object]: + return runpy.run_path(str(RUNNER), run_name=name) + + def test_browser_pass_retains_failure_process_termination_evidence(self) -> None: + """A browser-pass failure after identity capture must survive teardown as evidence.""" + + 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 ( + "browser_failure_type: str | None = None", + "browser_failure_type = type(exc).__name__", + '"failure_type": browser_failure_type', + '"browser_process_terminated": browser_process_terminated', + ): + with self.subTest(expected=expected): + self.assertIn(expected, browser_pass) + + def test_trial_preserves_failure_process_termination_evidence(self) -> None: + """The isolated trial must propagate failure teardown evidence after profile cleanup.""" + + namespace = self._namespace("agent_task_failure_process_termination_trial") + run_trial = namespace["_run_agent_task_trial"] + + def fail_after_shutdown(*_args: object, **_kwargs: object) -> dict[str, object]: + return { + "failure_type": "RuntimeError", + "browser_process_terminated": True, + } + + run_trial.__globals__["_run_agent_task_browser_pass"] = fail_after_shutdown + result = run_trial( + pathlib.Path("controlled-chrome"), + pathlib.Path("controlled-chromedriver"), + "http://127.0.0.1/controlled-fixture", + 11, + ) + + self.assertEqual(result["trial_number"], 11) + self.assertIs(result["passed"], False) + self.assertEqual(result["failure_type"], "RuntimeError") + self.assertIs(result["browser_process_terminated"], True) + self.assertIs(result["profile_cleaned"], True) + + def test_failed_trial_can_report_a_surviving_original_browser_process(self) -> None: + """Failure evidence must preserve a false result instead of inventing cleanup.""" + + namespace = self._namespace("agent_task_failure_process_survival_trial") + run_trial = namespace["_run_agent_task_trial"] + + def fail_with_surviving_process( + *_args: object, **_kwargs: object + ) -> dict[str, object]: + return { + "failure_type": "RuntimeError", + "browser_process_terminated": False, + } + + run_trial.__globals__["_run_agent_task_browser_pass"] = fail_with_surviving_process + result = run_trial( + pathlib.Path("controlled-chrome"), + pathlib.Path("controlled-chromedriver"), + "http://127.0.0.1/controlled-fixture", + 12, + ) + + self.assertIs(result["passed"], False) + self.assertIs(result["browser_process_terminated"], False) + self.assertIs(result["profile_cleaned"], True) + + +if __name__ == "__main__": + unittest.main()