test(browser): retain Agent Task process teardown evidence after failure

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())

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()