diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index 8b810caf4..dd9dbbde1 100644 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -5,7 +5,7 @@ W3C WebDriver HTTP protocol only to prove that a real Chrome for Testing build can load the controlled MV3 fixture and repeatedly exercise service-worker, content-script, storage, declarative-net-request, tabs, windows, scripting, -commands, side-panel, bookmarks, history, real browser-click, and +commands, side-panel, bookmarks, history, real-browser-click, and restart-persistence behavior. It also executes the controlled Agent Task fixture with extensions disabled in a fresh profile, locates the controlled action targets by exact browser-computed role/name evidence, performs real WebDriver @@ -16,7 +16,6 @@ from __future__ import annotations -import contextlib import hashlib import http.client import http.server @@ -127,8 +126,23 @@ def _json_request( if len(raw) > MAX_WEBDRIVER_RESPONSE_BYTES: raise RuntimeError("WebDriver response exceeded the bounded JSON limit") if response.status >= 400: - detail = raw.decode("utf-8", errors="replace") - raise RuntimeError(f"WebDriver HTTP {response.status}: {detail}") + try: + error_payload = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + error_payload = None + error_value = ( + error_payload.get("value") + if isinstance(error_payload, dict) + else None + ) + if ( + isinstance(error_value, dict) + and error_value.get("error") == "no such window" + ): + raise RuntimeError( + "WebDriver error: no such window: response details redacted" + ) + raise RuntimeError(f"WebDriver HTTP {response.status}") finally: connection.close() @@ -137,7 +151,11 @@ def _json_request( raise RuntimeError("WebDriver returned a non-object JSON payload") value = decoded.get("value") if isinstance(value, dict) and value.get("error"): - raise RuntimeError(f"WebDriver error: {value.get('error')}: {value.get('message')}") + if value.get("error") == "no such window": + raise RuntimeError( + "WebDriver error: no such window: response details redacted" + ) + raise RuntimeError("WebDriver returned an error response") return decoded @@ -151,7 +169,12 @@ def _wait_for_driver(driver_port: int) -> None: status = _json_request(driver_port, "GET", "/status", timeout=1.0) if status.get("value", {}).get("ready") is True: return - except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc: + except ( + OSError, + json.JSONDecodeError, + http.client.BadStatusLine, + http.client.IncompleteRead, + ) as exc: last_error = exc time.sleep(0.1) raise RuntimeError(f"ChromeDriver did not become ready: {last_error}") @@ -798,6 +821,64 @@ def _exercise_real_click(driver_port: int, session_id: str) -> str: return str(text) +def _validate_agent_task_submitted_state(state: object) -> None: + """Accept only the controlled submitted marker without echoing page state.""" + + if state != "submitted": + raise RuntimeError("Agent Task state post-condition failed") + + +def _delete_webdriver_session_bounded(driver_port: int, session_id: str) -> str | None: + """Delete one validated WebDriver session and retain only reviewed failure types.""" + + try: + _json_request( + driver_port, + "DELETE", + _webdriver_path(session_id, ""), + {}, + ) + except ( + OSError, + ValueError, + RuntimeError, + json.JSONDecodeError, + http.client.HTTPException, + ) as exc: + return type(exc).__name__ + return None + + +def _terminate_owned_process_bounded(process: Any) -> tuple[bool, str | None, bool]: + """Terminate one owned child under bounded waits and retain typed fallback evidence.""" + + try: + process.terminate() + except ProcessLookupError: + return True, None, False + except OSError as exc: + return False, type(exc).__name__, False + + try: + process.wait(timeout=PROCESS_EXIT_TIMEOUT_SECONDS) + return True, None, False + except subprocess.TimeoutExpired: + try: + process.kill() + except ProcessLookupError: + return True, None, True + except OSError as exc: + return False, type(exc).__name__, True + + try: + process.wait(timeout=PROCESS_EXIT_TIMEOUT_SECONDS) + return True, None, True + except subprocess.TimeoutExpired as exc: + return False, type(exc).__name__, True + except OSError as exc: + return False, type(exc).__name__, True + + def _run_browser_pass( chrome_bin: pathlib.Path, chromedriver_bin: pathlib.Path, @@ -809,6 +890,7 @@ def _run_browser_pass( driver_port = _free_loopback_port() session_id: str | None = None + primary_error: BaseException | None = None driver = subprocess.Popen( [str(chromedriver_bin), f"--port={driver_port}", "--allowed-ips=127.0.0.1"], stdout=subprocess.DEVNULL, @@ -893,21 +975,49 @@ def _run_browser_pass( "real-browser-click": click_result == "clicked", }, } + except BaseException as error: # noqa: BLE001 - re-raised unchanged after cleanup. + primary_error = error + raise finally: - if session_id is not None: - with contextlib.suppress(Exception): - _json_request( - driver_port, - "DELETE", - _webdriver_path(session_id, ""), - {}, + session_cleanup_failure_type = ( + _delete_webdriver_session_bounded(driver_port, session_id) + if session_id is not None + else None + ) + ( + driver_process_terminated, + driver_cleanup_failure_type, + driver_kill_fallback_used, + ) = _terminate_owned_process_bounded(driver) + if primary_error is not None: + if session_cleanup_failure_type is not None: + primary_error.add_note( + "WebDriver session cleanup also failed after the primary browser-pass " + f"failure: {session_cleanup_failure_type}" ) - driver.terminate() - try: - driver.wait(timeout=5) - except subprocess.TimeoutExpired: - driver.kill() - driver.wait(timeout=5) + if driver_cleanup_failure_type is not None or driver_process_terminated is not True: + cleanup_type = driver_cleanup_failure_type or "ProcessTerminationFailure" + primary_error.add_note( + "ChromeDriver process teardown also failed after the primary browser-pass " + f"failure: {cleanup_type}" + ) + elif session_cleanup_failure_type is not None: + cleanup_error = RuntimeError( + "WebDriver session cleanup failed after bounded process teardown" + ) + if driver_cleanup_failure_type is not None or driver_process_terminated is not True: + cleanup_type = driver_cleanup_failure_type or "ProcessTerminationFailure" + cleanup_error.add_note( + "ChromeDriver process teardown also failed: " + f"{cleanup_type}; kill_fallback_used={driver_kill_fallback_used}" + ) + raise cleanup_error + elif driver_cleanup_failure_type is not None or driver_process_terminated is not True: + cleanup_type = driver_cleanup_failure_type or "ProcessTerminationFailure" + raise RuntimeError( + "ChromeDriver process teardown failed after browser pass: " + f"{cleanup_type}; kill_fallback_used={driver_kill_fallback_used}" + ) def _run_restart_trial( @@ -1022,6 +1132,10 @@ def _run_agent_task_browser_pass( chromium_process_identities: tuple[tuple[int, int], ...] | None = None chromium_process_pre_shutdown_exit_count: int | None = None browser_failure_type: str | None = None + session_cleanup_failure_type: str | None = None + driver_process_terminated: bool | None = None + driver_cleanup_failure_type: str | None = None + driver_kill_fallback_used = False result: dict[str, Any] | None = None driver = subprocess.Popen( [str(chromedriver_bin), f"--port={driver_port}", "--allowed-ips=127.0.0.1"], @@ -1192,8 +1306,7 @@ def _run_agent_task_browser_pass( "GET", _element_command_path(session_id, result_element, "/text"), ).get("value") - if state != "submitted": - raise RuntimeError(f"Agent Task state post-condition failed: {state!r}") + _validate_agent_task_submitted_state(state) if text != AGENT_TASK_INPUT_VALUE: raise RuntimeError("Agent Task result did not match the synthetic typed value") structured_value_sha256 = _hash_agent_task_structured_value(text) @@ -1254,19 +1367,14 @@ def _run_agent_task_browser_pass( raise finally: if session_id is not None: - with contextlib.suppress(Exception): - _json_request( - driver_port, - "DELETE", - _webdriver_path(session_id, ""), - {}, - ) - driver.terminate() - try: - driver.wait(timeout=5) - except subprocess.TimeoutExpired: - driver.kill() - driver.wait(timeout=5) + session_cleanup_failure_type = _delete_webdriver_session_bounded( + driver_port, session_id + ) + ( + driver_process_terminated, + driver_cleanup_failure_type, + driver_kill_fallback_used, + ) = _terminate_owned_process_bounded(driver) if browser_process_id is None or browser_process_start_time_ticks is None: raise RuntimeError("Agent Task browser process identity was not captured") @@ -1279,11 +1387,30 @@ def _run_agent_task_browser_pass( chromium_process_set_terminated = _wait_for_linux_process_identity_set_exit( chromium_process_identities ) - if browser_failure_type is not None: + if ( + browser_failure_type is not None + or session_cleanup_failure_type is not None + or driver_cleanup_failure_type is not None + ): + primary_failure_type = browser_failure_type + if primary_failure_type is None and session_cleanup_failure_type is not None: + primary_failure_type = "WebDriverSessionCleanupError" + if primary_failure_type is None: + primary_failure_type = driver_cleanup_failure_type + if primary_failure_type is None: + raise RuntimeError("Agent Task failure evidence lost its primary type") failure_evidence: dict[str, Any] = { - "failure_type": browser_failure_type, + "failure_type": primary_failure_type, + "driver_process_terminated": driver_process_terminated, + "driver_kill_fallback_used": driver_kill_fallback_used, "browser_process_terminated": browser_process_terminated, } + if session_cleanup_failure_type is not None: + failure_evidence["session_cleanup_failure_type"] = session_cleanup_failure_type + if driver_cleanup_failure_type is not None and ( + browser_failure_type is not None or session_cleanup_failure_type is not None + ): + failure_evidence["cleanup_failure_type"] = driver_cleanup_failure_type if chromium_process_set_terminated is not None: failure_evidence["chromium_process_set_terminated"] = ( chromium_process_set_terminated @@ -1295,6 +1422,8 @@ def _run_agent_task_browser_pass( 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") + if driver_process_terminated is not True: + raise RuntimeError("Agent Task ChromeDriver process did not terminate") if not browser_process_terminated: raise RuntimeError("Agent Task browser process did not terminate") if not chromium_process_set_terminated: @@ -1475,6 +1604,11 @@ def _run_agent_task_forced_close_browser_pass( browser_process_id: int | None = None browser_process_start_time_ticks: int | None = None chromium_process_identities: tuple[tuple[int, int], ...] | None = None + browser_failure_type: str | None = None + session_cleanup_failure_type: str | None = None + driver_process_terminated: bool | None = None + driver_cleanup_failure_type: str | None = None + driver_kill_fallback_used = False result: dict[str, Any] | None = None driver = subprocess.Popen( [str(chromedriver_bin), f"--port={driver_port}", "--allowed-ips=127.0.0.1"], @@ -1619,39 +1753,75 @@ def _run_agent_task_forced_close_browser_pass( "forced_close_detected": forced_close_detected, "session_survived": True, } + 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): - _json_request( - driver_port, - "DELETE", - _webdriver_path(session_id, ""), - {}, - ) - driver.terminate() - try: - driver.wait(timeout=5) - except subprocess.TimeoutExpired: - driver.kill() - driver.wait(timeout=5) + session_cleanup_failure_type = _delete_webdriver_session_bounded( + driver_port, session_id + ) + ( + driver_process_terminated, + driver_cleanup_failure_type, + driver_kill_fallback_used, + ) = _terminate_owned_process_bounded(driver) 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 - ) + 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 + or session_cleanup_failure_type is not None + or driver_cleanup_failure_type is not None + ): + primary_failure_type = browser_failure_type + if primary_failure_type is None and session_cleanup_failure_type is not None: + primary_failure_type = "WebDriverSessionCleanupError" + if primary_failure_type is None: + primary_failure_type = driver_cleanup_failure_type + if primary_failure_type is None: + raise RuntimeError("Agent Task forced-close failure evidence lost its primary type") + failure_evidence: dict[str, Any] = { + "failure_type": primary_failure_type, + "driver_process_terminated": driver_process_terminated, + "driver_kill_fallback_used": driver_kill_fallback_used, + "browser_process_terminated": browser_process_terminated, + } + if session_cleanup_failure_type is not None: + failure_evidence["session_cleanup_failure_type"] = ( + session_cleanup_failure_type + ) + if driver_cleanup_failure_type is not None and ( + browser_failure_type is not None or session_cleanup_failure_type is not None + ): + failure_evidence["cleanup_failure_type"] = driver_cleanup_failure_type + 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 forced-close browser pass returned no result after shutdown") + if chromium_process_set_terminated is None: + raise RuntimeError("Agent Task forced-close Chromium process identities were not captured") + if driver_process_terminated is not True: + raise RuntimeError("Agent Task forced-close ChromeDriver process did not terminate") 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["driver_process_terminated"] = True + result["driver_kill_fallback_used"] = driver_kill_fallback_used result["browser_process_terminated"] = True result["chromium_process_set_terminated"] = True return result @@ -1705,6 +1875,56 @@ def _run_agent_task_forced_close_trial( } if result is None: raise RuntimeError("Agent Task forced-close 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 forced-close browser pass returned invalid failure evidence") + driver_process_terminated = result.get("driver_process_terminated") + if not isinstance(driver_process_terminated, bool): + raise RuntimeError("Agent Task forced-close browser pass returned invalid driver teardown evidence") + driver_kill_fallback_used = result.get("driver_kill_fallback_used") + if not isinstance(driver_kill_fallback_used, bool): + raise RuntimeError("Agent Task forced-close browser pass returned invalid driver fallback evidence") + browser_process_terminated = result.get("browser_process_terminated") + if not isinstance(browser_process_terminated, bool): + raise RuntimeError("Agent Task forced-close browser pass returned invalid teardown evidence") + failure_evidence: dict[str, Any] = { + "trial_number": trial_number, + "passed": False, + "failure_type": returned_failure_type, + "driver_process_terminated": driver_process_terminated, + "driver_kill_fallback_used": driver_kill_fallback_used, + "browser_process_terminated": browser_process_terminated, + "profile_cleaned": True, + "duration_ms": duration_ms, + } + if "session_cleanup_failure_type" in result: + session_cleanup_failure_type = result["session_cleanup_failure_type"] + if ( + not isinstance(session_cleanup_failure_type, str) + or not session_cleanup_failure_type + ): + raise RuntimeError( + "Agent Task forced-close browser pass returned invalid session cleanup failure evidence" + ) + failure_evidence["session_cleanup_failure_type"] = ( + session_cleanup_failure_type + ) + if "cleanup_failure_type" in result: + cleanup_failure_type = result["cleanup_failure_type"] + if not isinstance(cleanup_failure_type, str) or not cleanup_failure_type: + raise RuntimeError("Agent Task forced-close browser pass returned invalid cleanup failure evidence") + failure_evidence["cleanup_failure_type"] = cleanup_failure_type + 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 forced-close 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, @@ -1712,6 +1932,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"], + "driver_process_terminated": result["driver_process_terminated"], + "driver_kill_fallback_used": result["driver_kill_fallback_used"], "browser_process_terminated": result["browser_process_terminated"], "chromium_process_set_terminated": result["chromium_process_set_terminated"], "profile_cleaned": True, @@ -1892,11 +2114,6 @@ def main() -> int: and trial["browser_process_rss_bytes"] > 0 and isinstance(trial.get("chromium_process_count"), int) and 0 < trial["chromium_process_count"] <= MAX_BROWSER_PROCESS_TREE_SIZE - and isinstance(trial.get("chromium_process_pre_shutdown_exit_count"), int) - and not isinstance(trial["chromium_process_pre_shutdown_exit_count"], bool) - and 0 - <= trial["chromium_process_pre_shutdown_exit_count"] - < trial["chromium_process_count"] and isinstance(trial.get("chromium_process_set_rss_bytes"), int) and trial["chromium_process_set_rss_bytes"] > 0 and isinstance(trial.get("semantic_observation_bytes"), int) @@ -1919,6 +2136,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("driver_process_terminated") is True + and isinstance(trial.get("driver_kill_fallback_used"), bool) and trial.get("browser_process_terminated") is True and trial.get("chromium_process_set_terminated") is True and trial.get("profile_cleaned") is True @@ -1996,4 +2215,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 index 7aae4d475..a2ec5a0c5 100644 --- a/tests/test_agent_task_failure_process_termination_contract.py +++ b/tests/test_agent_task_failure_process_termination_contract.py @@ -26,7 +26,8 @@ def test_browser_pass_retains_failure_process_termination_evidence(self) -> None for expected in ( "browser_failure_type: str | None = None", "browser_failure_type = type(exc).__name__", - '"failure_type": browser_failure_type', + "primary_failure_type = browser_failure_type", + '"failure_type": primary_failure_type', '"browser_process_terminated": browser_process_terminated', ): with self.subTest(expected=expected): diff --git a/tests/test_agent_task_forced_close_process_termination_contract.py b/tests/test_agent_task_forced_close_process_termination_contract.py index 319227b98..d7a0e8871 100644 --- a/tests/test_agent_task_forced_close_process_termination_contract.py +++ b/tests/test_agent_task_forced_close_process_termination_contract.py @@ -4,6 +4,7 @@ import pathlib import runpy +import subprocess import unittest ROOT = pathlib.Path(__file__).resolve().parents[1] @@ -25,8 +26,11 @@ def test_forced_close_browser_pass_binds_and_waits_for_process_identities(self) "_read_linux_proc_stat_process_identity", "_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", + '"driver_process_terminated"', + '"driver_kill_fallback_used"', '"browser_process_terminated"', '"chromium_process_set_terminated"', ): @@ -51,6 +55,8 @@ def fake_browser_pass( "browser_version": namespace["PINNED_CHROME_VERSION"], "forced_close_detected": True, "session_survived": True, + "driver_process_terminated": True, + "driver_kill_fallback_used": False, "browser_process_terminated": False, "chromium_process_set_terminated": False, } @@ -62,22 +68,227 @@ def fake_browser_pass( "http://127.0.0.1/fixture", 1, ) + self.assertIs(result["driver_process_terminated"], True) + self.assertIs(result["driver_kill_fallback_used"], False) 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.""" + def test_forced_close_browser_failure_is_returned_after_teardown_waits(self) -> None: + """A reviewed browser failure after identity capture must not skip teardown proof.""" 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] + 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 ( - 'trial.get("browser_process_terminated") is True', - 'trial.get("chromium_process_set_terminated") is True', + "browser_failure_type", + "driver_cleanup_failure_type", + "driver_kill_fallback_used", + "except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:", + 'browser_failure_type = type(exc).__name__', + "failure_evidence", + '"driver_process_terminated": driver_process_terminated', + '"driver_kill_fallback_used": driver_kill_fallback_used', + '"browser_process_terminated": browser_process_terminated', + 'failure_evidence["chromium_process_set_terminated"]', ): with self.subTest(expected=expected): - self.assertIn(expected, gate) + 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) + + def test_forced_close_driver_shutdown_timeout_is_bounded_and_typed(self) -> None: + """A wedged ChromeDriver after SIGKILL must become failure evidence, not escape.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="forced_close_driver_shutdown_timeout_contract" + ) + shutdown = namespace["_terminate_owned_process_bounded"] + timeout_seconds = namespace["PROCESS_EXIT_TIMEOUT_SECONDS"] + + class WedgedProcess: + def __init__(self) -> None: + self.terminated = False + self.killed = False + self.wait_timeouts: list[float] = [] + + def terminate(self) -> None: + self.terminated = True + + def kill(self) -> None: + self.killed = True + + def wait(self, timeout: float) -> int: + self.wait_timeouts.append(timeout) + raise subprocess.TimeoutExpired("chromedriver", timeout) + + process = WedgedProcess() + terminated, failure_type, kill_fallback_used = shutdown(process) + + self.assertIs(process.terminated, True) + self.assertIs(process.killed, True) + self.assertEqual(process.wait_timeouts, [timeout_seconds, timeout_seconds]) + self.assertIs(terminated, False) + self.assertEqual(failure_type, "TimeoutExpired") + self.assertIs(kill_fallback_used, True) + + def test_forced_close_driver_shutdown_records_successful_kill_fallback(self) -> None: + """A successful SIGKILL fallback must remain explicit in cleanup evidence.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="forced_close_driver_kill_fallback_contract" + ) + shutdown = namespace["_terminate_owned_process_bounded"] + timeout_seconds = namespace["PROCESS_EXIT_TIMEOUT_SECONDS"] + + class KillRecoversProcess: + def __init__(self) -> None: + self.terminated = False + self.killed = False + self.wait_timeouts: list[float] = [] + + def terminate(self) -> None: + self.terminated = True + + def kill(self) -> None: + self.killed = True + + def wait(self, timeout: float) -> int: + self.wait_timeouts.append(timeout) + if len(self.wait_timeouts) == 1: + raise subprocess.TimeoutExpired("chromedriver", timeout) + return 0 + + process = KillRecoversProcess() + terminated, failure_type, kill_fallback_used = shutdown(process) + + self.assertIs(process.terminated, True) + self.assertIs(process.killed, True) + self.assertEqual(process.wait_timeouts, [timeout_seconds, timeout_seconds]) + self.assertIs(terminated, True) + self.assertIsNone(failure_type) + self.assertIs(kill_fallback_used, True) + + def test_forced_close_driver_shutdown_graceful_path_records_no_fallback(self) -> None: + """Graceful ChromeDriver shutdown must not report hard-kill fallback use.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="forced_close_driver_graceful_shutdown_contract" + ) + shutdown = namespace["_terminate_owned_process_bounded"] + timeout_seconds = namespace["PROCESS_EXIT_TIMEOUT_SECONDS"] + + class GracefulProcess: + def __init__(self) -> None: + self.terminated = False + self.killed = False + self.wait_timeouts: list[float] = [] + + def terminate(self) -> None: + self.terminated = True + + def kill(self) -> None: + self.killed = True + + def wait(self, timeout: float) -> int: + self.wait_timeouts.append(timeout) + return 0 + + process = GracefulProcess() + terminated, failure_type, kill_fallback_used = shutdown(process) + + self.assertIs(process.terminated, True) + self.assertIs(process.killed, False) + self.assertEqual(process.wait_timeouts, [timeout_seconds]) + self.assertIs(terminated, True) + self.assertIsNone(failure_type) + self.assertIs(kill_fallback_used, False) + + def test_forced_close_trial_preserves_driver_cleanup_failure_separately(self) -> None: + """Browser and driver-cleanup failure evidence must remain separately attributable.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="forced_close_driver_cleanup_trial_contract" + ) + 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 { + "failure_type": "RuntimeError", + "cleanup_failure_type": "TimeoutExpired", + "driver_process_terminated": False, + "driver_kill_fallback_used": True, + "browser_process_terminated": True, + "chromium_process_set_terminated": True, + } + + 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", + 2, + ) + + self.assertIs(result["passed"], False) + self.assertEqual(result["failure_type"], "RuntimeError") + self.assertEqual(result["cleanup_failure_type"], "TimeoutExpired") + self.assertIs(result["driver_process_terminated"], False) + self.assertIs(result["driver_kill_fallback_used"], True) + self.assertIs(result["browser_process_terminated"], True) + self.assertIs(result["chromium_process_set_terminated"], True) + self.assertIs(result["profile_cleaned"], True) + + def test_forced_close_trial_preserves_successful_kill_fallback_evidence(self) -> None: + """Successful forced-close trials must still say when ChromeDriver needed SIGKILL.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="forced_close_driver_kill_fallback_trial_contract" + ) + 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, + "driver_process_terminated": True, + "driver_kill_fallback_used": True, + "browser_process_terminated": True, + "chromium_process_set_terminated": True, + } + + 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", + 3, + ) + + self.assertIs(result["passed"], True) + self.assertIs(result["driver_process_terminated"], True) + self.assertIs(result["driver_kill_fallback_used"], True) + self.assertIs(result["browser_process_terminated"], True) + self.assertIs(result["chromium_process_set_terminated"], True) + self.assertIs(result["profile_cleaned"], True) if __name__ == "__main__": diff --git a/tests/test_agent_task_forced_close_session_cleanup_contract.py b/tests/test_agent_task_forced_close_session_cleanup_contract.py new file mode 100644 index 000000000..15e37a991 --- /dev/null +++ b/tests/test_agent_task_forced_close_session_cleanup_contract.py @@ -0,0 +1,120 @@ +"""Contract for truthful WebDriver session cleanup in the forced-close Agent Task lane.""" + +from __future__ import annotations + +import json +import pathlib +import runpy +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class AgentTaskForcedCloseSessionCleanupContractTests(unittest.TestCase): + """Require reviewed session-delete failures to remain explicit failure evidence.""" + + def test_session_delete_helper_is_bounded_typed_and_source_free(self) -> None: + """A reviewed WebDriver cleanup failure must return only its stable exception type.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="forced_close_session_cleanup_contract" + ) + cleanup = namespace["_delete_webdriver_session_bounded"] + original_request = cleanup.__globals__["_json_request"] + calls: list[tuple[int, str, str, dict[str, object]]] = [] + + def successful_request( + driver_port: int, + method: str, + path: str, + payload: dict[str, object], + ) -> dict[str, object]: + calls.append((driver_port, method, path, payload)) + return {"value": None} + + cleanup.__globals__["_json_request"] = successful_request + try: + self.assertIsNone(cleanup(9515, "session-1")) + finally: + cleanup.__globals__["_json_request"] = original_request + self.assertEqual(calls, [(9515, "DELETE", "/session/session-1", {})]) + + for exception in ( + OSError("raw-io-detail"), + ValueError("raw-value-detail"), + RuntimeError("raw-runtime-detail"), + json.JSONDecodeError("raw-json-detail", "x", 0), + ): + with self.subTest(exception_type=type(exception).__name__): + def failing_request(*_args: object, **_kwargs: object) -> dict[str, object]: + raise exception + + cleanup.__globals__["_json_request"] = failing_request + try: + failure_type = cleanup(9515, "session-1") + finally: + cleanup.__globals__["_json_request"] = original_request + self.assertEqual(failure_type, type(exception).__name__) + self.assertNotIn("raw-", failure_type) + + def test_forced_close_pass_does_not_suppress_session_cleanup_failure(self) -> None: + """The forced-close failure envelope must consume typed session cleanup evidence.""" + + 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.assertNotIn("contextlib.suppress(Exception)", browser_pass) + self.assertIn("session_cleanup_failure_type", browser_pass) + cleanup_call = browser_pass.index("_delete_webdriver_session_bounded(") + self.assertIn("driver_port, session_id", browser_pass[cleanup_call:cleanup_call + 160]) + self.assertIn('"session_cleanup_failure_type"', browser_pass) + self.assertIn('"WebDriverSessionCleanupError"', browser_pass) + + def test_trial_preserves_session_cleanup_failure_separately_from_driver_cleanup(self) -> None: + """Browser, session-delete, and driver-process failures must remain distinguishable.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="forced_close_session_cleanup_trial_contract" + ) + 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 { + "failure_type": "RuntimeError", + "session_cleanup_failure_type": "OSError", + "cleanup_failure_type": "TimeoutExpired", + "driver_process_terminated": False, + "driver_kill_fallback_used": True, + "browser_process_terminated": True, + "chromium_process_set_terminated": True, + } + + 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", + 4, + ) + + self.assertIs(result["passed"], False) + self.assertEqual(result["failure_type"], "RuntimeError") + self.assertEqual(result["session_cleanup_failure_type"], "OSError") + self.assertEqual(result["cleanup_failure_type"], "TimeoutExpired") + self.assertIs(result["driver_process_terminated"], False) + self.assertIs(result["driver_kill_fallback_used"], True) + self.assertIs(result["browser_process_terminated"], True) + self.assertIs(result["chromium_process_set_terminated"], True) + self.assertIs(result["profile_cleaned"], True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_agent_task_state_diagnostic_contract.py b/tests/test_agent_task_state_diagnostic_contract.py new file mode 100644 index 000000000..65400d869 --- /dev/null +++ b/tests/test_agent_task_state_diagnostic_contract.py @@ -0,0 +1,32 @@ +"""Regression contract for fail-closed, non-reflective Agent Task state diagnostics.""" + +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 AgentTaskStateDiagnosticContractTests(unittest.TestCase): + """Keep page-controlled state values out of runner diagnostics.""" + + def test_post_condition_failure_does_not_reflect_page_controlled_state(self) -> None: + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_state_diagnostic_contract") + validate = namespace["_validate_agent_task_submitted_state"] + hostile_state = "rejected" + + with self.assertRaisesRegex( + RuntimeError, + r"^Agent Task state post-condition failed$", + ) as captured: + validate(hostile_state) + + self.assertNotIn(hostile_state, str(captured.exception)) + validate("submitted") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_chromedriver_error_diagnostic_contract.py b/tests/test_chromedriver_error_diagnostic_contract.py new file mode 100644 index 000000000..f9f0f9734 --- /dev/null +++ b/tests/test_chromedriver_error_diagnostic_contract.py @@ -0,0 +1,75 @@ +"""Regression tests for credential-safe ChromeDriver error diagnostics.""" + +from __future__ import annotations + +import http.server +import pathlib +import runpy +import threading +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" +SECRET_MARKER = "buyer-secret-marker-must-not-reach-ci" + + +class _ErrorResponseHandler(http.server.BaseHTTPRequestHandler): + """Serve deterministic hostile ChromeDriver-shaped error responses.""" + + response_status = 403 + response_body = ( + b'{"value":{"error":"unknown error","message":"' + + SECRET_MARKER.encode("ascii") + + b'"}}' + ) + + def do_GET(self) -> None: # noqa: N802 - stdlib handler contract. + self.send_response(self.response_status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(self.response_body))) + self.end_headers() + self.wfile.write(self.response_body) + + def log_message(self, _format: str, *args: object) -> None: + """Keep the hostile marker out of test-server logging.""" + + +class ChromeDriverErrorDiagnosticContractTests(unittest.TestCase): + """ChromeDriver-controlled response bytes must not be reflected into CI errors.""" + + def _request_against(self, *, status: int) -> RuntimeError: + namespace = runpy.run_path(str(RUNNER), run_name="chromedriver_error_diagnostic_contract") + json_request = namespace["_json_request"] + + class Handler(_ErrorResponseHandler): + response_status = status + + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + with self.assertRaises(RuntimeError) as captured: + json_request(int(server.server_port), "GET", "/status") + return captured.exception + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2.0) + + def test_http_error_does_not_reflect_response_body(self) -> None: + """An HTTP error may retain its status but never ChromeDriver-controlled detail.""" + + error = self._request_against(status=403) + self.assertIn("403", str(error)) + self.assertNotIn(SECRET_MARKER, str(error)) + + def test_webdriver_error_does_not_reflect_response_message(self) -> None: + """A 2xx WebDriver error object must remain fail-closed without its raw message.""" + + error = self._request_against(status=200) + self.assertIn("WebDriver", str(error)) + self.assertNotIn(SECRET_MARKER, str(error)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_chromedriver_startup_exception_contract.py b/tests/test_chromedriver_startup_exception_contract.py new file mode 100644 index 000000000..931b0de6f --- /dev/null +++ b/tests/test_chromedriver_startup_exception_contract.py @@ -0,0 +1,54 @@ +"""Fail-closed exception contract for bounded ChromeDriver startup probing.""" + +from __future__ import annotations + +import http.client +import pathlib +import runpy +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class ChromeDriverStartupExceptionContractTests(unittest.TestCase): + """Keep recoverable startup transport faults separate from terminal failures.""" + + def test_runner_startup_retries_transient_incomplete_response(self) -> None: + """A truncated startup response may be retried within the existing deadline.""" + + namespace = runpy.run_path(str(RUNNER), run_name="chromedriver_incomplete_startup_response") + wait_for_driver = namespace["_wait_for_driver"] + attempts = [0] + + def truncated_then_ready(*_args: object, **_kwargs: object) -> dict[str, object]: + attempts[0] += 1 + if attempts[0] == 1: + raise http.client.IncompleteRead(b'{"value":', 20) + return {"value": {"ready": True}} + + wait_for_driver.__globals__["_json_request"] = truncated_then_ready + wait_for_driver(9515) + self.assertEqual(attempts[0], 2) + + def test_runner_startup_does_not_retry_terminal_runtime_failure(self) -> None: + """A terminal WebDriver/runtime failure must fail closed before a later success.""" + + namespace = runpy.run_path(str(RUNNER), run_name="chromedriver_terminal_startup_failure") + wait_for_driver = namespace["_wait_for_driver"] + attempts = [0] + + def terminal_then_ready(*_args: object, **_kwargs: object) -> dict[str, object]: + attempts[0] += 1 + if attempts[0] == 1: + raise RuntimeError("WebDriver HTTP 403: forbidden") + return {"value": {"ready": True}} + + wait_for_driver.__globals__["_json_request"] = terminal_then_ready + with self.assertRaisesRegex(RuntimeError, "HTTP 403"): + wait_for_driver(9515) + self.assertEqual(attempts[0], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mv3_compatibility_contract.py b/tests/test_mv3_compatibility_contract.py index 10872ddac..71734b0c8 100644 --- a/tests/test_mv3_compatibility_contract.py +++ b/tests/test_mv3_compatibility_contract.py @@ -2,6 +2,7 @@ from __future__ import annotations +import http.client import json import pathlib import runpy @@ -120,6 +121,46 @@ def test_runner_transport_cannot_follow_dynamic_url_schemes(self) -> None: self.assertNotIn("urllib.request", runner) self.assertNotIn("urllib.error", runner) + def test_runner_cleanup_cannot_suppress_untyped_failures(self) -> None: + """Cleanup failures must stay typed evidence instead of becoming false-green runs.""" + + runner = RUNNER.read_text(encoding="utf-8") + self.assertNotIn("contextlib.suppress(Exception)", runner) + self.assertIn("_delete_webdriver_session_bounded", runner) + self.assertIn("_terminate_owned_process_bounded", runner) + + def test_runner_session_cleanup_classifies_http_protocol_failure(self) -> None: + """Malformed ChromeDriver HTTP during cleanup must remain typed evidence.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_cleanup_http_failure") + delete_session = namespace["_delete_webdriver_session_bounded"] + + def fail_with_bad_status(*_args: object, **_kwargs: object) -> dict[str, object]: + raise http.client.BadStatusLine("malformed status line") + + delete_session.__globals__["_json_request"] = fail_with_bad_status + self.assertEqual( + delete_session(9515, "controlled-session"), + "BadStatusLine", + ) + + def test_runner_startup_retries_http_protocol_failure(self) -> None: + """A transient malformed ChromeDriver startup response must be retried boundedly.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_startup_http_failure") + wait_for_driver = namespace["_wait_for_driver"] + attempts = [0] + + def transient_bad_status(*_args: object, **_kwargs: object) -> dict[str, object]: + attempts[0] += 1 + if attempts[0] == 1: + raise http.client.BadStatusLine("malformed status line") + return {"value": {"ready": True}} + + wait_for_driver.__globals__["_json_request"] = transient_bad_status + wait_for_driver(9515) + self.assertEqual(attempts[0], 2) + def test_runner_accepts_real_chromedriver_element_ids_without_path_injection(self) -> None: """ChromeDriver dotted element IDs must work while path syntax stays fail-closed."""