diff --git a/AGENTS.md b/AGENTS.md index 6f747c38e..be69de4f8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -21,6 +21,9 @@ For every change: Do not bypass required checks, branch protection, or any review authority actually required by current GitHub rules or an explicit operationally satisfiable OriginWeave/CWL governance rule. Waiting checks are not permission to weaken tests; continue with a non-conflicting next task. +- ChromeDriver diagnostic drains must continuously discard bytes, accept existing text-mode process doubles by immediate UTF-8 replacement encoding, and silently stop on an unrecognized chunk; only the reviewed startup reason may survive. Shared ChromeDriver launch owns `--verbose` and never writes a diagnostic log file. +- A `session not created` result may race the asynchronous ChromeDriver drain. Wait only for a reviewed startup-reason event or bounded handoff expiry before choosing a closed reason; unreviewed output, process exit, and arbitrary sleeps are not acceptable handoff signals. + ## Work-conserving autonomous maintenance **A completed action is an intermediate state**, not an implicit end of a maintenance invocation. “One bounded slice” means **one write-active slice at a time**, not one slice, pull request, RCA, check, review request, documentation update, or merge per run. diff --git a/CHANGELOG.md b/CHANGELOG.md index 61b8208f3..600c522d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to OriginWeave are documented in this file. The format follo ## [Unreleased] +- ChromeDriver session-creation failures now wait for one bounded process-diagnostic handoff before selecting a closed startup reason, preventing a delayed reviewed sandbox marker from being recorded as `unknown` without retaining output or waiting for process exit. +- ChromeDriver's shared diagnostic drain now accepts existing text-mode process doubles and discards unrecognized chunks without a background thread error; the shared launch retains verbose process evidence without a log file. +- ChromeDriver `session not created` responses now retain a typed `session_not_created` failure and only the allowlisted `sandbox_unavailable` or `unknown` startup reason; raw driver-controlled response text remains excluded from browser-crash evidence. +- Browser-crash trials no longer disable Chromium's sandbox. Rejected session startup remains one failed attempt with driver and temporary-profile cleanup, without an unsandboxed retry; live pinned-browser acceptance is still required. +- Browser-crash failure evidence now preserves the first causal browser failure when session cleanup or ChromeDriver teardown also fails, retains secondary cleanup only as bounded exception-type fields, and excludes raw exception text from the emitted artifact. Cleanup-only failures remain fail-closed as primary failures. + ### Added - Kept the inherited protocol-failure cleanup checks executable after shared-deadline integration by observing the correct cleanup path for ordinary and forced-close trials; all failure, cleanup and diagnostic-redaction assertions remain intact. @@ -42,6 +48,9 @@ All notable changes to OriginWeave are documented in this file. The format follo ### Changed +- Kept ChromeDriver startup diagnosis fail-closed when verbose output races session creation: only a reviewed startup marker can complete the bounded handoff early, while ordinary process output waits for the existing expiry before retaining `unknown`. + +- Controlled browser-crash compatibility evidence now credits a crash only after the exact PID/start-time identity is signalled through a revalidated Linux pidfd and that same pidfd becomes readable within the bounded deadline; generic WebDriver transport failures no longer substitute for process-termination proof, while sampled Chromium process-set teardown remains a separate recovery boundary and the pidfd runtime contract remains mandatory on Linux CI. - Separated logical origin authority from resolved network destination authority; an origin grant no longer implies permission to connect to every resolver result. - Separated resolved-address authorization from direct transport evidence; an approved IP now becomes a usable stream only after the operating system reports the exact requested IP and port. - Separated exact TCP peer proof from authenticated TLS service identity; an observed peer becomes an authenticated HTTPS stream only after explicit-root, fixed-time, SAN-bound WebPKI verification over that same stream. diff --git a/CLAUDE.md b/CLAUDE.md index ab08bc16f..c3bef2033 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -11,4 +11,6 @@ Additional constraints: - Do not merge logical origin, destination authorization, direct TCP peer proof, TLS service identity, proxy routing, or HTTP resource policy into one ambient authority. - Do not add hostname reconnect, proxy-environment inheritance, dangerous certificate-verifier hooks, Common Name fallback, TLS 0-RTT, key logging, or secret extraction to a production TLS path. - Keep changes bounded to one product gap and preserve modular crate boundaries. +- ChromeDriver process output is a redaction boundary: support text-mode test doubles by encoding and discard unrecognized chunks without retaining them; emit only the closed diagnostic reason. +- For concurrent session failure, use a reviewed startup-reason event or bounded handoff expiry rather than arbitrary output, a sleep, or process-exit wait before reading the closed reason. - Never claim a test, benchmark, browser integration, TLS identity, GPU execution, release, or merge succeeded without current exact-head evidence. diff --git a/docs/doctoring.md b/docs/doctoring.md index 0ffbe0317..4d4b5fd2b 100644 --- a/docs/doctoring.md +++ b/docs/doctoring.md @@ -124,6 +124,16 @@ At #147 `af1b98ba377c73b88baa9633e2232e7f76e76f36`, all six parent review-eviden The existing test now specifies each lane's actual observer, return value and exact arguments. Ordinary trials keep the individual `False` result. Forced-close trials inject `(False, False)` from the shared observer and require `(321, 654, ((321, 654),))`, preserving root-only capture rather than inventing a complete descendant set. Every existing failed-trial, profile-cleanup, live-root, driver-termination, error-type and redaction assertion remains. No production code, deadline, retry, failed denominator or coverage gate changes. This repair does not address the separate ordinary-pass two-deadline finding or prove real Linux/pinned-Chromium acceptance. +### Browser-crash sandbox conformance + +PR #148 adopts the current #147 parent `3dff28d9bf2dd27b72507e39979d51b8bf140fb4` by ordinary merge, retaining the crash/pidfd delta and all six inherited cleanup-evidence regressions. A new regression executes the actual outer crash trial and intercepts its session-start request: before the repair it fails because that request disables the Chromium sandbox. Removing that one crash-lane argument restores the existing sandbox invariant without introducing a configuration switch or fallback path. + +The same regression requires one failed startup attempt, driver reaping and actual temporary-profile removal. It preserves browser executable selection and never credits the failed startup as a successful crash trial. The existing PID-safe signal/exit observers, sampled-process checks and trial denominator are unchanged. Other launch lanes retain their separate owner repairs; this child must not be interpreted as complete runner-wide sandbox integration. Mocked launch requests prove the control-flow contract, not that a real Chromium binary started with its sandbox active. Linux pidfd tests and exact-head pinned-Chromium compatibility remain separate acceptance evidence; macOS skips are not passes. + +ChromeDriver session creation is also an evidence boundary. A structured `session not created` response is retained as a typed `session_not_created` failure. Only the reviewed `No usable sandbox` diagnostic maps to `sandbox_unavailable`; every other driver-controlled message maps to `unknown`. Raw response text, executable/profile paths and arbitrary diagnostics remain excluded, so the next pinned-browser run can distinguish the sandbox-helper case without admitting untrusted ChromeDriver prose into CI evidence. This classification does not install the helper, retry startup, disable the sandbox or make cleanup equivalent to browser success. + +Supplemental verification used the existing Colima Linux kernel `6.8.0-117-generic` and Python `3.12.3`, without installing dependencies or changing VM configuration. All 247 Python contracts execute there with no skips, including the three real pidfd cases skipped on macOS: killed-but-unreaped child, non-terminating signal and stale identity. Initial host-path discovery failed because this VM does not expose the host worktree. The first streamed archive then added AppleDouble `._*.rs` files, causing two TLS source-read errors. A fresh export of only Git-tracked paths with `COPYFILE_DISABLE=1 tar --no-xattrs --no-acls --no-fflags` removes that packaging artifact at its producer; no test filter or source exception was added. Source and regression-file SHA-256 hashes match across hosts. This proves the supplemental Linux contracts, not pinned-Chromium execution, hosted approval or release acceptance. + ## References Amazon Web Services. (n.d.). *Set up the Amazon EKS Pod Identity Agent*. Retrieved August 6, 2026, from https://docs.aws.amazon.com/eks/latest/userguide/pod-id-agent-setup.html diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index 5fd3b29c2..c2636e729 100644 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -23,6 +23,8 @@ import math import os import pathlib +import select +import signal import socket import string import subprocess @@ -43,6 +45,7 @@ STARTUP_TIMEOUT_SECONDS = 20.0 FIXTURE_TIMEOUT_SECONDS = 20.0 PROCESS_EXIT_TIMEOUT_SECONDS = 5.0 +DIAGNOSTIC_HANDOFF_TIMEOUT_SECONDS = 0.25 MAX_WEBDRIVER_RESPONSE_BYTES = 1_048_576 MAX_PROC_STATUS_CHARACTERS = 65_536 MAX_PROC_STAT_CHARACTERS = 65_536 @@ -56,6 +59,62 @@ PATH_TOKEN_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_.") +class _WebDriverNoSuchWindowError(RuntimeError): + """Identify reviewed ChromeDriver no-such-window evidence without masking other failures.""" + + +class _WebDriverSessionNotCreatedError(RuntimeError): + """Retain only closed ChromeDriver startup evidence from a rejected session.""" + + error_code = "session_not_created" + + def __init__(self, startup_reason: str) -> None: + """Build one redacted typed failure with an allowlisted startup reason.""" + + if startup_reason not in {"sandbox_unavailable", "unknown"}: + raise ValueError("unsupported WebDriver session startup reason") + self.startup_reason = startup_reason + super().__init__("WebDriver error: session not created: response details redacted") + + +class _ChromeDriverStartupDiagnostic: + """Retain only a closed startup reason while continuously discarding process output.""" + + __slots__ = ("_marker_index", "_reviewed_reason", "startup_reason") + _MARKER = b"no usable sandbox" + + def __init__(self) -> None: + """Start with no reviewed process-level startup reason.""" + + self._marker_index = 0 + self._reviewed_reason = threading.Event() + self.startup_reason = "unknown" + + def feed(self, chunk: bytes) -> None: + """Scan one output chunk without retaining raw ChromeDriver-controlled bytes.""" + + if not isinstance(chunk, bytes): + raise TypeError("ChromeDriver diagnostic chunks must be bytes") + if self.startup_reason == "sandbox_unavailable": + return + for raw_byte in chunk: + byte = raw_byte + 32 if 65 <= raw_byte <= 90 else raw_byte + if byte == self._MARKER[self._marker_index]: + self._marker_index += 1 + if self._marker_index == len(self._MARKER): + self.startup_reason = "sandbox_unavailable" + self._reviewed_reason.set() + self._marker_index = 0 + return + else: + self._marker_index = 1 if byte == self._MARKER[0] else 0 + + def wait_for_observation(self) -> None: + """Bound the handoff from asynchronous process draining to session classification.""" + + self._reviewed_reason.wait(timeout=DIAGNOSTIC_HANDOFF_TIMEOUT_SECONDS) + + class QuietFixtureHandler(http.server.SimpleHTTPRequestHandler): """Serve only the controlled local fixture without noisy access logging.""" @@ -95,6 +154,15 @@ def _webdriver_path(session_id: str, suffix: str) -> str: return f"/session/{safe_session}{suffix}" +def _classify_webdriver_session_startup_reason(error_value: dict[str, Any]) -> str: + """Map reviewed ChromeDriver startup text to a closed credential-safe reason.""" + + message = error_value.get("message") + if isinstance(message, str) and "no usable sandbox" in message.casefold(): + return "sandbox_unavailable" + return "unknown" + + def _json_request( driver_port: int, method: str, @@ -139,9 +207,16 @@ def _json_request( isinstance(error_value, dict) and error_value.get("error") == "no such window" ): - raise RuntimeError( + raise _WebDriverNoSuchWindowError( "WebDriver error: no such window: response details redacted" ) + if ( + isinstance(error_value, dict) + and error_value.get("error") == "session not created" + ): + raise _WebDriverSessionNotCreatedError( + _classify_webdriver_session_startup_reason(error_value) + ) raise RuntimeError(f"WebDriver HTTP {response.status}") finally: connection.close() @@ -152,9 +227,13 @@ def _json_request( value = decoded.get("value") if isinstance(value, dict) and value.get("error"): if value.get("error") == "no such window": - raise RuntimeError( + raise _WebDriverNoSuchWindowError( "WebDriver error: no such window: response details redacted" ) + if value.get("error") == "session not created": + raise _WebDriverSessionNotCreatedError( + _classify_webdriver_session_startup_reason(value) + ) raise RuntimeError("WebDriver returned an error response") return decoded @@ -180,6 +259,73 @@ def _wait_for_driver(driver_port: int) -> None: raise RuntimeError(f"ChromeDriver did not become ready: {last_error}") +def _drain_chromedriver_diagnostics( + stream: Any, + diagnostic: _ChromeDriverStartupDiagnostic, +) -> None: + """Continuously drain ChromeDriver output while retaining only reviewed reason state.""" + + while True: + chunk = stream.read(8_192) + if not chunk: + return + if isinstance(chunk, str): + chunk = chunk.encode("utf-8", "replace") + elif not isinstance(chunk, bytes): + return + diagnostic.feed(chunk) + + +def _start_chromedriver( + chromedriver_bin: pathlib.Path, + driver_port: int, +) -> tuple[subprocess.Popen[Any], _ChromeDriverStartupDiagnostic]: + """Start one local ChromeDriver and continuously drain its credential-bearing output.""" + + diagnostic = _ChromeDriverStartupDiagnostic() + driver = subprocess.Popen( + [ + str(chromedriver_bin), + f"--port={driver_port}", + "--allowed-ips=127.0.0.1", + "--verbose", + ], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + if driver.stdout is None: + driver.terminate() + driver.wait(timeout=PROCESS_EXIT_TIMEOUT_SECONDS) + raise RuntimeError("ChromeDriver diagnostic pipe was unavailable") + threading.Thread( + target=_drain_chromedriver_diagnostics, + args=(driver.stdout, diagnostic), + daemon=True, + ).start() + return driver, diagnostic + + +def _create_chromedriver_session( + driver_port: int, + payload: dict[str, Any], + diagnostic: _ChromeDriverStartupDiagnostic, +) -> dict[str, Any]: + """Create one session while allowing only reviewed process startup evidence to refine errors.""" + + _wait_for_driver(driver_port) + try: + return _json_request(driver_port, "POST", "/session", payload) + except _WebDriverSessionNotCreatedError as error: + if error.startup_reason == "unknown": + diagnostic.wait_for_observation() + if ( + error.startup_reason == "unknown" + and diagnostic.startup_reason == "sandbox_unavailable" + ): + raise _WebDriverSessionNotCreatedError("sandbox_unavailable") from None + raise + + def _execute(driver_port: int, session_id: str, script: str) -> Any: """Run fixture-only JavaScript through the test WebDriver session.""" @@ -479,7 +625,7 @@ def _read_linux_proc_stat_process_identity(process_id: int) -> tuple[int, int] | try: with stat_path.open("r", encoding="utf-8", errors="strict") as stat_file: stat_text = stat_file.read(MAX_PROC_STAT_CHARACTERS + 1) - except FileNotFoundError: + except (FileNotFoundError, ProcessLookupError): return None if len(stat_text) > MAX_PROC_STAT_CHARACTERS: raise RuntimeError("Linux proc stat exceeded the bounded text limit") @@ -489,6 +635,111 @@ def _read_linux_proc_stat_process_identity(process_id: int) -> tuple[int, int] | return identity +def _signal_linux_process_identity( + process_identity: tuple[int, int], + signal_number: int, +) -> bool: + """Signal only one exact Linux PID/start-time identity through a pidfd.""" + + if not isinstance(process_identity, tuple) or len(process_identity) != 2: + raise ValueError("invalid Linux process identity") + process_id, start_time_ticks = process_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 isinstance(signal_number, bool) or not isinstance(signal_number, int) or signal_number <= 0: + raise ValueError("invalid Linux process signal") + + expected_identity = (process_id, start_time_ticks) + if _read_linux_proc_stat_process_identity(process_id) != expected_identity: + return False + pidfd_open = getattr(os, "pidfd_open", None) + pidfd_send_signal = getattr(signal, "pidfd_send_signal", None) + if not callable(pidfd_open) or not callable(pidfd_send_signal): + raise RuntimeError("Linux pidfd signalling is unavailable") + try: + pidfd = pidfd_open(process_id, 0) + except ProcessLookupError: + return False + try: + if _read_linux_proc_stat_process_identity(process_id) != expected_identity: + return False + try: + pidfd_send_signal(pidfd, signal_number) + except ProcessLookupError: + return False + return True + finally: + os.close(pidfd) + + +def _signal_and_wait_for_linux_process_identity_termination( + process_identity: tuple[int, int], + signal_number: int, + *, + timeout_seconds: float = PROCESS_EXIT_TIMEOUT_SECONDS, +) -> bool: + """Signal one exact Linux identity and await termination on that same pidfd.""" + + if not isinstance(process_identity, tuple) or len(process_identity) != 2: + raise ValueError("invalid Linux process identity") + process_id, start_time_ticks = process_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 ( + isinstance(signal_number, bool) + or not isinstance(signal_number, int) + or signal_number <= 0 + ): + raise ValueError("invalid Linux process signal") + 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 termination timeout") + + expected_identity = (process_id, start_time_ticks) + if _read_linux_proc_stat_process_identity(process_id) != expected_identity: + return False + pidfd_open = getattr(os, "pidfd_open", None) + pidfd_send_signal = getattr(signal, "pidfd_send_signal", None) + if not callable(pidfd_open) or not callable(pidfd_send_signal): + raise RuntimeError("Linux pidfd signalling is unavailable") + try: + pidfd = pidfd_open(process_id, 0) + except ProcessLookupError: + return False + try: + if _read_linux_proc_stat_process_identity(process_id) != expected_identity: + return False + try: + pidfd_send_signal(pidfd, signal_number) + except ProcessLookupError: + return False + readable, _writable, _exceptional = select.select( + [pidfd], + [], + [], + float(timeout_seconds), + ) + return bool(readable) + finally: + os.close(pidfd) + + def _wait_for_linux_process_identity_exit( process_id: int, start_time_ticks: int, @@ -961,18 +1212,10 @@ 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, - stderr=subprocess.STDOUT, - text=True, - ) + driver, startup_diagnostic = _start_chromedriver(chromedriver_bin, driver_port) try: - _wait_for_driver(driver_port) - session = _json_request( + session = _create_chromedriver_session( driver_port, - "POST", - "/session", { "capabilities": { "alwaysMatch": { @@ -986,7 +1229,6 @@ def _run_browser_pass( "--disable-component-update", "--disable-sync", "--disable-dev-shm-usage", - "--no-sandbox", f"--user-data-dir={profile_dir}", f"--disable-extensions-except={FIXTURE}", f"--load-extension={FIXTURE}", @@ -995,6 +1237,7 @@ def _run_browser_pass( } } }, + startup_diagnostic, ).get("value", {}) if not isinstance(session, dict): raise RuntimeError("ChromeDriver session response is malformed") @@ -1215,18 +1458,10 @@ def _run_agent_task_browser_pass( 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"], - stdout=subprocess.DEVNULL, - stderr=subprocess.STDOUT, - text=True, - ) + driver, startup_diagnostic = _start_chromedriver(chromedriver_bin, driver_port) try: - _wait_for_driver(driver_port) - session = _json_request( + session = _create_chromedriver_session( driver_port, - "POST", - "/session", { "capabilities": { "alwaysMatch": { @@ -1244,7 +1479,6 @@ def _run_agent_task_browser_pass( "--disable-component-update", "--disable-sync", "--disable-dev-shm-usage", - "--no-sandbox", "--disable-extensions", f"--user-data-dir={profile_dir}", ], @@ -1252,6 +1486,7 @@ def _run_agent_task_browser_pass( } } }, + startup_diagnostic, ).get("value", {}) if not isinstance(session, dict): raise RuntimeError("ChromeDriver Agent Task session response is malformed") @@ -1689,18 +1924,10 @@ def _run_agent_task_forced_close_browser_pass( 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"], - stdout=subprocess.DEVNULL, - stderr=subprocess.STDOUT, - text=True, - ) + driver, startup_diagnostic = _start_chromedriver(chromedriver_bin, driver_port) try: - _wait_for_driver(driver_port) - session = _json_request( + session = _create_chromedriver_session( driver_port, - "POST", - "/session", { "capabilities": { "alwaysMatch": { @@ -1714,7 +1941,6 @@ def _run_agent_task_forced_close_browser_pass( "--disable-component-update", "--disable-sync", "--disable-dev-shm-usage", - "--no-sandbox", "--disable-extensions", f"--user-data-dir={profile_dir}", ], @@ -1722,6 +1948,7 @@ def _run_agent_task_forced_close_browser_pass( } } }, + startup_diagnostic, ).get("value", {}) if not isinstance(session, dict): raise RuntimeError("ChromeDriver forced-close session response is malformed") @@ -2028,6 +2255,361 @@ def _run_agent_task_forced_close_trial( } +def _cleanup_crashed_browser_session(driver_port: int, session_id: str | None) -> None: + """Delete a crash session while ignoring only reviewed post-crash transport loss.""" + + if session_id is None: + return + try: + _json_request( + driver_port, + "DELETE", + _webdriver_path(session_id, ""), + {}, + ) + except ( + OSError, + _WebDriverNoSuchWindowError, + json.JSONDecodeError, + http.client.IncompleteRead, + ): + return + + +def _stop_crashed_driver(driver: subprocess.Popen[Any]) -> None: + """Reap ChromeDriver without re-signalling a child that already exited.""" + + if driver.poll() is not None: + driver.wait(timeout=5) + return + driver.terminate() + try: + driver.wait(timeout=5) + except subprocess.TimeoutExpired: + driver.kill() + driver.wait(timeout=5) + + +def _run_agent_task_browser_crash_browser_pass( + chrome_bin: pathlib.Path, + chromedriver_bin: pathlib.Path, + fixture_url: str, + profile_dir: str, +) -> dict[str, Any]: + """Kill one exact browser root and prove crash detection plus sampled teardown.""" + + _require_pristine_agent_task_profile(profile_dir) + 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 + browser_version: str | None = None + browser_process_crash_detected = False + driver, startup_diagnostic = _start_chromedriver(chromedriver_bin, driver_port) + try: + session = _create_chromedriver_session( + driver_port, + { + "capabilities": { + "alwaysMatch": { + "browserName": "chrome", + "goog:chromeOptions": { + "binary": str(chrome_bin), + "prefs": { + "credentials_enable_service": False, + "profile.password_manager_enabled": False, + }, + "args": [ + "--headless=new", + "--no-first-run", + "--disable-default-apps", + "--disable-component-update", + "--disable-sync", + "--disable-dev-shm-usage", + "--disable-extensions", + f"--user-data-dir={profile_dir}", + ], + }, + } + } + }, + startup_diagnostic, + ).get("value", {}) + if not isinstance(session, dict): + raise RuntimeError("ChromeDriver browser-crash session response is malformed") + raw_session_id = session.get("sessionId") + capabilities = session.get("capabilities", {}) + if not isinstance(raw_session_id, str): + raise RuntimeError("ChromeDriver did not return a browser-crash session id") + if not isinstance(capabilities, dict): + raise RuntimeError("ChromeDriver browser-crash 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 browser-crash 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 browser-crash process id") + browser_process_identity = _read_linux_proc_stat_process_identity(browser_process_id) + if browser_process_identity is None: + raise RuntimeError("Agent Task browser-crash process identity disappeared") + browser_process_start_time_ticks = browser_process_identity[1] + + _json_request( + driver_port, + "POST", + _webdriver_path(session_id, "/url"), + {"url": fixture_url}, + ) + loaded_url = _json_request( + driver_port, + "GET", + _webdriver_path(session_id, "/url"), + ).get("value") + if loaded_url != fixture_url: + raise RuntimeError("Agent Task browser-crash 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_identity, + ) + ) + if not _signal_and_wait_for_linux_process_identity_termination( + browser_process_identity, + signal.SIGKILL, + ): + raise RuntimeError( + "Agent Task browser process was not observed terminated after crash signal" + ) + browser_process_crash_detected = True + finally: + try: + _cleanup_crashed_browser_session(driver_port, session_id) + finally: + _stop_crashed_driver(driver) + + if ( + browser_process_id is None + or browser_process_start_time_ticks is None + or chromium_process_identities is None + or browser_version is None + ): + raise RuntimeError("Agent Task browser-crash teardown identities were not captured") + browser_process_terminated, chromium_process_set_terminated = ( + _wait_for_linux_process_teardown( + browser_process_id, + browser_process_start_time_ticks, + chromium_process_identities, + ) + ) + if not browser_process_crash_detected: + raise RuntimeError("Agent Task browser-process crash was not detected") + if not browser_process_terminated: + raise RuntimeError("Agent Task browser-crash root process did not terminate") + if not chromium_process_set_terminated: + raise RuntimeError("Agent Task browser-crash Chromium process set did not terminate") + return { + "browser_version": browser_version, + "browser_process_crash_detected": True, + "browser_process_terminated": True, + "chromium_process_set_terminated": True, + } + + +def _classify_agent_task_browser_crash_reason(error: BaseException) -> str: + """Map crash failures onto a closed reason vocabulary without retaining messages.""" + + if isinstance(error, _WebDriverSessionNotCreatedError): + return error.error_code + if isinstance(error, subprocess.TimeoutExpired): + return "timeout" + if isinstance(error, json.JSONDecodeError): + return "invalid_json" + if isinstance(error, ValueError): + return "invalid_value" + if isinstance(error, OSError): + return "os_error" + return "runtime_error" + + +def _classify_agent_task_browser_crash_stage(error: BaseException) -> str: + """Classify one crash failure from bounded traceback structure and pass state.""" + + traceback_cursor = error.__traceback__ + pass_locals: dict[str, Any] | None = None + function_names: set[str] = set() + while traceback_cursor is not None: + frame = traceback_cursor.tb_frame + function_name = frame.f_code.co_name + function_names.add(function_name) + if function_name == "_run_agent_task_browser_crash_browser_pass": + pass_locals = dict(frame.f_locals) + traceback_cursor = traceback_cursor.tb_next + + if "_wait_for_driver" in function_names: + return "driver_ready" + if "_cleanup_crashed_browser_session" in function_names: + return "session_cleanup" + if "_stop_crashed_driver" in function_names: + return "driver_teardown" + if "_signal_and_wait_for_linux_process_identity_termination" in function_names: + return "crash_signal" + if "_wait_for_linux_process_teardown" in function_names: + return "post_crash_teardown" + if function_names.intersection( + { + "_snapshot_linux_process_evidence", + "_discover_linux_process_tree_ids", + "_read_linux_process_identity_set", + } + ): + return "process_tree_capture" + + if pass_locals is None: + return "browser_pass" + if "driver" not in pass_locals: + return "driver_start" + if pass_locals.get("session_id") is None: + return "session_create" + if pass_locals.get("browser_process_id") is None: + return "session_identity" + if pass_locals.get("browser_process_start_time_ticks") is None: + return "browser_identity" + if pass_locals.get("chromium_process_identities") is None: + return "fixture_navigation" + if pass_locals.get("browser_process_crash_detected") is not True: + return "crash_signal" + return "post_crash_teardown" + + +def _partition_agent_task_browser_crash_failure( + error: BaseException, +) -> tuple[BaseException, str | None, str | None]: + """Preserve the first crash failure while retaining typed secondary cleanup evidence.""" + + primary_error = error + session_cleanup_failure_type: str | None = None + driver_cleanup_failure_type: str | None = None + visited: set[int] = set() + while id(primary_error) not in visited: + visited.add(id(primary_error)) + stage = _classify_agent_task_browser_crash_stage(primary_error) + context = primary_error.__context__ + if context is None: + break + if stage == "driver_teardown": + driver_cleanup_failure_type = type(primary_error).__name__ + primary_error = context + continue + if stage == "session_cleanup": + session_cleanup_failure_type = type(primary_error).__name__ + primary_error = context + continue + break + return primary_error, session_cleanup_failure_type, driver_cleanup_failure_type + + +def _run_agent_task_browser_crash_trial( + chrome_bin: pathlib.Path, + chromedriver_bin: pathlib.Path, + fixture_url: str, + trial_number: int, +) -> dict[str, Any]: + """Run one isolated browser-root crash trial and retain cleanup evidence.""" + + trial_started = time.monotonic() + profile_path: pathlib.Path + result: dict[str, Any] | None = None + failure_type: str | None = None + failure_stage: str | None = None + reason_code: str | None = None + startup_reason: str | None = None + session_cleanup_failure_type: str | None = None + cleanup_failure_type: str | None = None + with tempfile.TemporaryDirectory( + prefix=f"originweave-agent-task-browser-crash-{trial_number}-" + ) as profile_dir: + profile_path = pathlib.Path(profile_dir) + try: + result = _run_agent_task_browser_crash_browser_pass( + chrome_bin, + chromedriver_bin, + fixture_url, + profile_dir, + ) + except ( + OSError, + ValueError, + RuntimeError, + json.JSONDecodeError, + subprocess.TimeoutExpired, + ) as exc: + ( + primary_error, + session_cleanup_failure_type, + cleanup_failure_type, + ) = _partition_agent_task_browser_crash_failure(exc) + failure_type = type(primary_error).__name__ + failure_stage = _classify_agent_task_browser_crash_stage(primary_error) + reason_code = _classify_agent_task_browser_crash_reason(primary_error) + if isinstance(primary_error, _WebDriverSessionNotCreatedError): + startup_reason = primary_error.startup_reason + profile_cleaned = not profile_path.exists() + if not profile_cleaned: + raise RuntimeError( + f"Agent Task browser-crash profile cleanup failed in trial {trial_number}" + ) + + duration_ms = round((time.monotonic() - trial_started) * 1000) + if failure_type is not None: + if failure_stage is None or reason_code is None: + raise RuntimeError("Agent Task browser-crash failure classification was incomplete") + failure_evidence: dict[str, Any] = { + "trial_number": trial_number, + "passed": False, + "failure_type": failure_type, + "failure_stage": failure_stage, + "reason_code": reason_code, + "profile_cleaned": True, + "duration_ms": duration_ms, + } + if session_cleanup_failure_type is not None: + failure_evidence["session_cleanup_failure_type"] = ( + session_cleanup_failure_type + ) + if cleanup_failure_type is not None: + failure_evidence["cleanup_failure_type"] = cleanup_failure_type + if startup_reason is not None: + failure_evidence["startup_reason"] = startup_reason + return failure_evidence + if result is None: + raise RuntimeError("Agent Task browser-crash browser pass returned no result") + return { + "trial_number": trial_number, + "passed": True, + "browser_version": result["browser_version"], + "browser_process_crash_detected": result["browser_process_crash_detected"], + "browser_process_terminated": result["browser_process_terminated"], + "chromium_process_set_terminated": result["chromium_process_set_terminated"], + "profile_cleaned": True, + "duration_ms": duration_ms, + } + + def _start_fixture_server( directory: pathlib.Path, ) -> tuple[http.server.ThreadingHTTPServer, threading.Thread]: @@ -2163,6 +2745,26 @@ def main() -> int: } ) + browser_crash_trials: list[dict[str, Any]] = [] + for trial_number in range(1, AGENT_TASK_REPEATABILITY_TRIALS + 1): + try: + browser_crash_trials.append( + _run_agent_task_browser_crash_trial( + chrome_bin, + chromedriver_bin, + agent_task_url, + trial_number, + ) + ) + except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc: + browser_crash_trials.append( + { + "trial_number": trial_number, + "passed": False, + "failure_type": type(exc).__name__, + } + ) + agent_task_successful_trials = sum( 1 for trial in agent_task_trials if trial.get("passed") is True ) @@ -2234,6 +2836,20 @@ def main() -> int: for trial in forced_close_trials if trial.get("passed") is True ) + browser_crash_successful_trials = sum( + 1 for trial in browser_crash_trials if trial.get("passed") is True + ) + browser_crash_profiles_cleaned = all( + trial.get("profile_cleaned") is True for trial in browser_crash_trials + ) + browser_crash_surfaces_complete = all( + trial.get("browser_process_crash_detected") 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 browser_crash_trials + if trial.get("passed") is True + ) evidence = { "chrome_version": PINNED_CHROME_VERSION, @@ -2262,6 +2878,12 @@ def main() -> int: "profiles_cleaned": forced_close_profiles_cleaned, "trial_results": forced_close_trials, }, + "browser_crash": { + "repeatability_trials": AGENT_TASK_REPEATABILITY_TRIALS, + "successful_trials": browser_crash_successful_trials, + "profiles_cleaned": browser_crash_profiles_cleaned, + "trial_results": browser_crash_trials, + }, }, "duration_ms": round((time.monotonic() - started) * 1000), } @@ -2298,6 +2920,17 @@ def main() -> int: f"{forced_close_successful_trials}/{AGENT_TASK_REPEATABILITY_TRIALS} " "trials passed" ) + if not browser_crash_profiles_cleaned: + raise RuntimeError("Agent Task browser-crash profile cleanup gate failed") + if ( + browser_crash_successful_trials != AGENT_TASK_REPEATABILITY_TRIALS + or not browser_crash_surfaces_complete + ): + raise RuntimeError( + "Agent Task browser-crash recovery gate failed: " + f"{browser_crash_successful_trials}/{AGENT_TASK_REPEATABILITY_TRIALS} " + "trials passed" + ) return 0 finally: _stop_fixture_server(agent_task_server, agent_task_thread) diff --git a/tests/test_agent_task_browser_crash_cleanup_runtime_contract.py b/tests/test_agent_task_browser_crash_cleanup_runtime_contract.py new file mode 100644 index 000000000..ba5ba260c --- /dev/null +++ b/tests/test_agent_task_browser_crash_cleanup_runtime_contract.py @@ -0,0 +1,198 @@ +"""Fail-closed contract for unexpected browser-crash cleanup runtime failures.""" + +from __future__ import annotations + +import pathlib +import runpy +import unittest +from unittest import mock + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class AgentTaskBrowserCrashCleanupRuntimeContractTests(unittest.TestCase): + """Keep unknown WebDriver/runtime cleanup failures visible after a browser crash.""" + + def test_unknown_runtime_error_is_not_suppressed(self) -> None: + """Only reviewed post-crash transport loss may be converted to cleanup success.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="agent_task_browser_crash_cleanup_runtime_contract" + ) + cleanup_session = namespace["_cleanup_crashed_browser_session"] + + def unexpected_runtime_failure(*_args: object, **_kwargs: object) -> object: + raise RuntimeError("unexpected WebDriver protocol failure") + + cleanup_session.__globals__["_json_request"] = unexpected_runtime_failure + with self.assertRaisesRegex(RuntimeError, "unexpected WebDriver protocol failure"): + cleanup_session(9222, "session-1") + + def test_primary_failure_survives_secondary_session_cleanup_failure(self) -> None: + """Session cleanup diagnostics must not replace the first browser failure stage.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="agent_task_browser_crash_primary_failure_contract" + ) + run_pass = namespace["_run_agent_task_browser_crash_browser_pass"] + pinned_version = namespace["PINNED_CHROME_VERSION"] + driver = mock.Mock() + driver.poll.return_value = 0 + + def request( + _driver_port: int, + method: str, + path: str, + _payload: object, + ) -> dict[str, object]: + if method == "POST" and path == "/session": + return { + "value": { + "sessionId": "session-1", + "capabilities": { + "browserVersion": pinned_version, + "goog:processID": 777, + }, + } + } + if method == "POST" and path.endswith("/url"): + raise RuntimeError("primary fixture navigation failure") + if method == "DELETE": + raise RuntimeError("secondary cleanup failure") + raise AssertionError(f"unexpected WebDriver request: {method} {path}") + + with ( + mock.patch.dict( + run_pass.__globals__, + { + "_free_loopback_port": lambda: 9222, + "_wait_for_driver": lambda _port: None, + "_json_request": request, + "_read_linux_proc_stat_process_identity": lambda _pid: (777, 42), + }, + ), + mock.patch.object( + run_pass.__globals__["subprocess"], + "Popen", + return_value=driver, + ), + ): + result = namespace["_run_agent_task_browser_crash_trial"]( + pathlib.Path("/controlled/chrome"), + pathlib.Path("/controlled/chromedriver"), + "http://127.0.0.1/agent-task", + 1, + ) + + self.assertFalse(result["passed"]) + self.assertEqual(result["failure_type"], "RuntimeError") + self.assertEqual(result["failure_stage"], "fixture_navigation") + self.assertEqual(result["reason_code"], "runtime_error") + self.assertEqual(result["session_cleanup_failure_type"], "RuntimeError") + self.assertNotIn("primary fixture navigation failure", repr(result)) + self.assertNotIn("secondary cleanup failure", repr(result)) + self.assertTrue(result["profile_cleaned"]) + driver.wait.assert_called_once_with(timeout=5) + + def test_primary_failure_survives_secondary_driver_teardown_failure(self) -> None: + """Driver teardown diagnostics must not replace the first browser failure stage.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="agent_task_browser_crash_driver_failure_contract" + ) + run_pass = namespace["_run_agent_task_browser_crash_browser_pass"] + pinned_version = namespace["PINNED_CHROME_VERSION"] + driver = mock.Mock() + driver.poll.return_value = 0 + driver.wait.side_effect = RuntimeError("secondary driver teardown failure") + + def request( + _driver_port: int, + method: str, + path: str, + _payload: object, + ) -> dict[str, object]: + if method == "POST" and path == "/session": + return { + "value": { + "sessionId": "session-1", + "capabilities": { + "browserVersion": pinned_version, + "goog:processID": 777, + }, + } + } + if method == "POST" and path.endswith("/url"): + raise RuntimeError("primary fixture navigation failure") + if method == "DELETE": + return {"value": None} + raise AssertionError(f"unexpected WebDriver request: {method} {path}") + + with ( + mock.patch.dict( + run_pass.__globals__, + { + "_free_loopback_port": lambda: 9222, + "_wait_for_driver": lambda _port: None, + "_json_request": request, + "_read_linux_proc_stat_process_identity": lambda _pid: (777, 42), + }, + ), + mock.patch.object( + run_pass.__globals__["subprocess"], + "Popen", + return_value=driver, + ), + ): + result = namespace["_run_agent_task_browser_crash_trial"]( + pathlib.Path("/controlled/chrome"), + pathlib.Path("/controlled/chromedriver"), + "http://127.0.0.1/agent-task", + 1, + ) + + self.assertFalse(result["passed"]) + self.assertEqual(result["failure_type"], "RuntimeError") + self.assertEqual(result["failure_stage"], "fixture_navigation") + self.assertEqual(result["reason_code"], "runtime_error") + self.assertEqual(result["cleanup_failure_type"], "RuntimeError") + self.assertNotIn("primary fixture navigation failure", repr(result)) + self.assertNotIn("secondary driver teardown failure", repr(result)) + self.assertTrue(result["profile_cleaned"]) + driver.wait.assert_called_once_with(timeout=5) + + def test_cleanup_only_failure_remains_primary(self) -> None: + """Do not demote cleanup failure when no earlier browser failure exists.""" + + namespace = runpy.run_path( + str(RUNNER), run_name="agent_task_browser_crash_cleanup_only_contract" + ) + cleanup_session = namespace["_cleanup_crashed_browser_session"] + partition_failure = namespace["_partition_agent_task_browser_crash_failure"] + classify_stage = namespace["_classify_agent_task_browser_crash_stage"] + + def unexpected_runtime_failure(*_args: object, **_kwargs: object) -> object: + raise RuntimeError("cleanup-only failure") + + cleanup_session.__globals__["_json_request"] = unexpected_runtime_failure + captured_error: RuntimeError | None = None + try: + cleanup_session(9222, "session-1") + except RuntimeError as error: + captured_error = error + primary_error, session_failure_type, driver_failure_type = partition_failure( + error + ) + else: + self.fail("cleanup-only RuntimeError was unexpectedly suppressed") + + self.assertIsNotNone(captured_error) + self.assertIs(primary_error, captured_error) + self.assertEqual(classify_stage(primary_error), "session_cleanup") + self.assertIsNone(session_failure_type) + self.assertIsNone(driver_failure_type) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_agent_task_browser_crash_exact_exit_detection.py b/tests/test_agent_task_browser_crash_exact_exit_detection.py new file mode 100644 index 000000000..8b4590829 --- /dev/null +++ b/tests/test_agent_task_browser_crash_exact_exit_detection.py @@ -0,0 +1,154 @@ +"""Contract for exact browser-root termination evidence before crash credit.""" + +from __future__ import annotations + +import contextlib +import pathlib +import runpy +import signal +import subprocess +import sys +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class AgentTaskBrowserCrashExactExitDetectionContractTests(unittest.TestCase): + """Prevent transport failures or unreaped zombies from faking crash evidence.""" + + def test_crash_detection_uses_exact_pidfd_termination_observation(self) -> None: + """Signal and observe termination through one exact kernel process handle.""" + + runner = RUNNER.read_text(encoding="utf-8") + start = runner.index("def _run_agent_task_browser_crash_browser_pass(") + end = runner.index("\ndef _run_agent_task_browser_crash_trial(", start) + browser_pass = runner[start:end] + + termination_call = browser_pass.index( + "_signal_and_wait_for_linux_process_identity_termination(" + ) + crash_credit = browser_pass.index( + "browser_process_crash_detected = True", termination_call + ) + + self.assertLess(termination_call, crash_credit) + self.assertNotIn( + "_wait_for_linux_process_identity_exit(", + browser_pass[termination_call:crash_credit], + ) + self.assertNotIn( + "except (OSError, RuntimeError, json.JSONDecodeError):", + browser_pass[termination_call:crash_credit], + ) + + @unittest.skipUnless(sys.platform.startswith("linux"), "requires Linux pidfd") + def test_pidfd_termination_observes_killed_unreaped_child(self) -> None: + """Kernel termination evidence must not require the parent to reap the child.""" + + namespace = runpy.run_path( + str(RUNNER), + run_name="agent_task_browser_crash_exact_termination_contract", + ) + signal_and_wait = namespace[ + "_signal_and_wait_for_linux_process_identity_termination" + ] + read_identity = namespace["_read_linux_proc_stat_process_identity"] + + child = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + identity = read_identity(child.pid) + self.assertIsNotNone(identity) + assert identity is not None + + self.assertTrue( + signal_and_wait(identity, signal.SIGKILL, timeout_seconds=1.0) + ) + self.assertEqual( + read_identity(child.pid), + identity, + "the killed child should still be observable as unreaped procfs identity", + ) + finally: + with contextlib.suppress(ProcessLookupError): + child.kill() + child.wait(timeout=5) + + @unittest.skipUnless(sys.platform.startswith("linux"), "requires Linux pidfd") + def test_pidfd_termination_does_not_credit_signal_delivery_as_exit(self) -> None: + """A successfully delivered non-terminating signal is not termination evidence.""" + + namespace = runpy.run_path( + str(RUNNER), + run_name="agent_task_browser_crash_nonterminating_signal_contract", + ) + signal_and_wait = namespace[ + "_signal_and_wait_for_linux_process_identity_termination" + ] + read_identity = namespace["_read_linux_proc_stat_process_identity"] + + child = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + identity = read_identity(child.pid) + self.assertIsNotNone(identity) + assert identity is not None + + self.assertFalse( + signal_and_wait(identity, signal.SIGCONT, timeout_seconds=0.05) + ) + self.assertIsNone( + child.poll(), + "signal delivery without process exit must not be credited as termination", + ) + finally: + with contextlib.suppress(ProcessLookupError): + child.kill() + child.wait(timeout=5) + + @unittest.skipUnless(sys.platform.startswith("linux"), "requires Linux pidfd") + def test_pidfd_termination_refuses_stale_identity_without_signalling(self) -> None: + """A stale PID/start-time identity must never signal the current process owner.""" + + namespace = runpy.run_path( + str(RUNNER), + run_name="agent_task_browser_crash_stale_identity_contract", + ) + signal_and_wait = namespace[ + "_signal_and_wait_for_linux_process_identity_termination" + ] + read_identity = namespace["_read_linux_proc_stat_process_identity"] + + child = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + identity = read_identity(child.pid) + self.assertIsNotNone(identity) + assert identity is not None + stale_identity = (identity[0], identity[1] + 1) + + self.assertFalse( + signal_and_wait(stale_identity, signal.SIGKILL, timeout_seconds=0.1) + ) + self.assertIsNone( + child.poll(), + "stale identity validation must happen before any signal is delivered", + ) + finally: + with contextlib.suppress(ProcessLookupError): + child.kill() + child.wait(timeout=5) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_agent_task_browser_crash_recovery_contract.py b/tests/test_agent_task_browser_crash_recovery_contract.py new file mode 100644 index 000000000..aaafc0237 --- /dev/null +++ b/tests/test_agent_task_browser_crash_recovery_contract.py @@ -0,0 +1,383 @@ +"""Contract for PID-safe browser-process crash evidence in the Agent Task lane.""" + +from __future__ import annotations + +import http.client +import json +import os +import pathlib +import runpy +import signal +import subprocess +import unittest +from unittest import mock + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" +ORIGINAL_OS_CLOSE = os.close +ORIGINAL_PIDFD_OPEN = getattr(os, "pidfd_open", None) +ORIGINAL_PIDFD_SEND_SIGNAL = getattr(signal, "pidfd_send_signal", None) + + +class AgentTaskBrowserCrashRecoveryContractTests(unittest.TestCase): + """Require controlled browser-process interruption without PID-reuse races.""" + + def test_runner_exposes_pidfd_crash_boundary(self) -> None: + """The Linux browser crash probe must use a PID-safe signalling boundary.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_contract") + for expected in ( + "_signal_linux_process_identity", + "_run_agent_task_browser_crash_browser_pass", + "_run_agent_task_browser_crash_trial", + ): + with self.subTest(expected=expected): + self.assertIn(expected, namespace) + + def test_signal_boundary_rejects_pid_reuse_before_open(self) -> None: + """A reused PID must never receive the crash signal.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_reuse") + signal_identity = namespace["_signal_linux_process_identity"] + opened: list[int] = [] + signalled: list[tuple[int, int]] = [] + + signal_identity.__globals__["_read_linux_proc_stat_process_identity"] = ( + lambda _process_id: (777, 99) + ) + os_module = signal_identity.__globals__["os"] + signal_module = signal_identity.__globals__["signal"] + with ( + mock.patch.object( + os_module, + "pidfd_open", + side_effect=lambda process_id, _flags=0: opened.append(process_id) or 12, + create=True, + ), + mock.patch.object( + signal_module, + "pidfd_send_signal", + side_effect=lambda pidfd, sig, *_args, **_kwargs: signalled.append( + (pidfd, sig) + ), + create=True, + ), + mock.patch.object(os_module, "close", side_effect=lambda _fd: None), + ): + self.assertFalse(signal_identity((777, 42), signal.SIGKILL)) + self.assertEqual(opened, []) + self.assertEqual(signalled, []) + + def test_signal_boundary_rechecks_identity_after_pidfd_open(self) -> None: + """The process identity must still match after the race-free handle is opened.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_post_open") + signal_identity = namespace["_signal_linux_process_identity"] + identities = iter(((777, 42), (777, 99))) + signalled: list[tuple[int, int]] = [] + closed: list[int] = [] + + signal_identity.__globals__["_read_linux_proc_stat_process_identity"] = ( + lambda _process_id: next(identities) + ) + os_module = signal_identity.__globals__["os"] + signal_module = signal_identity.__globals__["signal"] + with ( + mock.patch.object( + os_module, + "pidfd_open", + side_effect=lambda _process_id, _flags=0: 12, + create=True, + ), + mock.patch.object( + signal_module, + "pidfd_send_signal", + side_effect=lambda pidfd, sig, *_args, **_kwargs: signalled.append( + (pidfd, sig) + ), + create=True, + ), + mock.patch.object(os_module, "close", side_effect=closed.append), + ): + self.assertFalse(signal_identity((777, 42), signal.SIGKILL)) + self.assertEqual(signalled, []) + self.assertEqual(closed, [12]) + + def test_signal_boundary_targets_only_exact_open_identity(self) -> None: + """Exact identity proof must send one signal through the opened pidfd and close it.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_exact") + signal_identity = namespace["_signal_linux_process_identity"] + signalled: list[tuple[int, int]] = [] + closed: list[int] = [] + + signal_identity.__globals__["_read_linux_proc_stat_process_identity"] = ( + lambda _process_id: (777, 42) + ) + os_module = signal_identity.__globals__["os"] + signal_module = signal_identity.__globals__["signal"] + with ( + mock.patch.object( + os_module, + "pidfd_open", + side_effect=lambda _process_id, _flags=0: 12, + create=True, + ), + mock.patch.object( + signal_module, + "pidfd_send_signal", + side_effect=lambda pidfd, sig, *_args, **_kwargs: signalled.append( + (pidfd, sig) + ), + create=True, + ), + mock.patch.object(os_module, "close", side_effect=closed.append), + ): + self.assertTrue(signal_identity((777, 42), signal.SIGKILL)) + self.assertEqual(signalled, [(12, signal.SIGKILL)]) + self.assertEqual(closed, [12]) + + def test_signal_boundary_handles_exit_before_pidfd_signal(self) -> None: + """A target that exits after pidfd open must become a bounded not-signalled result.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_esrch") + signal_identity = namespace["_signal_linux_process_identity"] + closed: list[int] = [] + + signal_identity.__globals__["_read_linux_proc_stat_process_identity"] = ( + lambda _process_id: (777, 42) + ) + + def exited_before_signal(*_args: object, **_kwargs: object) -> None: + raise ProcessLookupError("process exited before pidfd signal") + + os_module = signal_identity.__globals__["os"] + signal_module = signal_identity.__globals__["signal"] + with ( + mock.patch.object( + os_module, + "pidfd_open", + side_effect=lambda _process_id, _flags=0: 12, + create=True, + ), + mock.patch.object( + signal_module, + "pidfd_send_signal", + side_effect=exited_before_signal, + create=True, + ), + mock.patch.object(os_module, "close", side_effect=closed.append), + ): + self.assertFalse(signal_identity((777, 42), signal.SIGKILL)) + self.assertEqual(closed, [12]) + + def test_proc_stat_identity_treats_read_time_esrch_as_process_exit(self) -> None: + """A process disappearing while procfs is read must become bounded absence.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_proc_esrch") + read_identity = namespace["_read_linux_proc_stat_process_identity"] + + class VanishedStat: + def __enter__(self) -> "VanishedStat": + return self + + def __exit__(self, *_args: object) -> None: + return None + + def read(self, _limit: int) -> str: + raise ProcessLookupError("process exited during proc stat read") + + with mock.patch.object(pathlib.Path, "open", return_value=VanishedStat()): + self.assertIsNone(read_identity(777)) + + def test_crash_driver_cleanup_is_idempotent_after_driver_exit(self) -> None: + """A browser crash may end ChromeDriver before cleanup without a second signal.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_cleanup") + cleanup = namespace["_stop_crashed_driver"] + events: list[object] = [] + + class ExitedDriver: + def poll(self) -> int: + events.append("poll") + return 0 + + def terminate(self) -> None: + raise AssertionError("already-exited ChromeDriver must not be re-signalled") + + def wait(self, *, timeout: int) -> int: + events.append(("wait", timeout)) + return 0 + + def kill(self) -> None: + raise AssertionError("already-exited ChromeDriver must not be killed") + + cleanup(ExitedDriver()) + self.assertEqual(events, ["poll", ("wait", 5)]) + + def test_crash_trial_records_driver_teardown_timeout_as_failed_trial(self) -> None: + """A bounded ChromeDriver teardown timeout must fail one trial without aborting the run.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_timeout") + run_trial = namespace["_run_agent_task_browser_crash_trial"] + + def driver_teardown_timeout(*_args: object, **_kwargs: object) -> dict[str, object]: + raise subprocess.TimeoutExpired(cmd="chromedriver", timeout=5) + + run_trial.__globals__["_run_agent_task_browser_crash_browser_pass"] = ( + driver_teardown_timeout + ) + result = run_trial( + pathlib.Path("/controlled/chrome"), + pathlib.Path("/controlled/chromedriver"), + "http://127.0.0.1/agent-task", + 1, + ) + + self.assertFalse(result["passed"]) + self.assertEqual(result["failure_type"], "TimeoutExpired") + self.assertTrue(result["profile_cleaned"]) + + def test_crash_reason_codes_are_closed_and_type_stable(self) -> None: + """Diagnostic reason codes must distinguish reviewed exception classes without messages.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_reasons") + classify = namespace["_classify_agent_task_browser_crash_reason"] + cases = ( + (subprocess.TimeoutExpired(cmd="chromedriver", timeout=5), "timeout"), + (json.JSONDecodeError("redacted", "{}", 0), "invalid_json"), + (ValueError("redacted"), "invalid_value"), + (OSError("redacted"), "os_error"), + (RuntimeError("redacted"), "runtime_error"), + ) + for error, expected in cases: + with self.subTest(error_type=type(error).__name__): + self.assertEqual(classify(error), expected) + + def test_crash_startup_keeps_sandbox_and_fails_without_fallback(self) -> None: + """A rejected sandboxed session must fail once, reap its driver and remove its profile.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_sandbox") + run_pass = namespace["_run_agent_task_browser_crash_browser_pass"] + session_error = namespace["_WebDriverSessionNotCreatedError"] + request = mock.Mock(side_effect=session_error("sandbox_unavailable")) + driver = mock.Mock() + driver.poll.return_value = 0 + with ( + mock.patch.dict( + run_pass.__globals__, + { + "_free_loopback_port": lambda: 9222, + "_wait_for_driver": lambda _port: None, + "_json_request": request, + }, + ), + mock.patch.object(subprocess, "Popen", return_value=driver) as launch, + ): + result = namespace["_run_agent_task_browser_crash_trial"]( + pathlib.Path("/controlled/chrome"), + pathlib.Path("/controlled/chromedriver"), + "http://127.0.0.1/agent-task", + 1, + ) + + self.assertFalse(result["passed"]) + self.assertEqual(result["failure_type"], "_WebDriverSessionNotCreatedError") + self.assertEqual(result["failure_stage"], "session_create") + self.assertEqual(result["reason_code"], "session_not_created") + self.assertEqual(result["startup_reason"], "sandbox_unavailable") + self.assertNotIn("failure_message", result) + self.assertNotIn("response details", repr(result)) + self.assertTrue(result["profile_cleaned"]) + launch.assert_called_once() + driver.wait.assert_called_once_with(timeout=5) + request.assert_called_once() + self.assertEqual(request.call_args.args[:3], (9222, "POST", "/session")) + options = request.call_args.args[3]["capabilities"]["alwaysMatch"]["goog:chromeOptions"] + self.assertEqual(options["binary"], "/controlled/chrome") + profile_args = [arg for arg in options["args"] if arg.startswith("--user-data-dir=")] + self.assertEqual(len(profile_args), 1) + self.assertFalse(pathlib.Path(profile_args[0].split("=", 1)[1]).exists()) + self.assertNotIn("--no-sandbox", options["args"]) + + def test_crash_session_cleanup_ignores_only_reviewed_transport_failures(self) -> None: + """Expected post-crash transport loss is bounded, while programming failures propagate.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_session_cleanup") + self.assertIn("_cleanup_crashed_browser_session", namespace) + cleanup_session = namespace["_cleanup_crashed_browser_session"] + calls: list[tuple[object, ...]] = [] + + def expected_transport_failure(*args: object, **_kwargs: object) -> object: + calls.append(args) + raise OSError("browser transport is already gone") + + cleanup_session.__globals__["_json_request"] = expected_transport_failure + cleanup_session(9222, "session-1") + self.assertEqual(len(calls), 1) + + def incomplete_post_crash_response(*_args: object, **_kwargs: object) -> object: + raise http.client.IncompleteRead(b'{"value":') + + cleanup_session.__globals__["_json_request"] = incomplete_post_crash_response + cleanup_session(9222, "session-1") + + def unexpected_programming_failure(*_args: object, **_kwargs: object) -> object: + raise AssertionError("unexpected cleanup defect") + + cleanup_session.__globals__["_json_request"] = unexpected_programming_failure + with self.assertRaisesRegex(AssertionError, "unexpected cleanup defect"): + cleanup_session(9222, "session-1") + + def test_crash_session_cleanup_without_session_is_a_noop(self) -> None: + """No session identifier means cleanup has no remote operation to attempt.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_browser_crash_no_session") + self.assertIn("_cleanup_crashed_browser_session", namespace) + cleanup_session = namespace["_cleanup_crashed_browser_session"] + + def unexpected_request(*_args: object, **_kwargs: object) -> object: + raise AssertionError("cleanup must not call WebDriver without a session") + + cleanup_session.__globals__["_json_request"] = unexpected_request + cleanup_session(9222, None) + + def test_crash_lane_binds_sampled_process_set_to_exact_root_identity(self) -> None: + """Crash sampling must preserve the prerequisite root-identity integrity contract.""" + + runner = RUNNER.read_text(encoding="utf-8") + crash_lane = runner.split( + "def _run_agent_task_browser_crash_browser_pass", 1 + )[1].split("def _run_agent_task_browser_crash_trial", 1)[0] + self.assertIn( + "required_root_identity=browser_process_identity", + crash_lane, + ) + + def test_crash_lane_is_required_for_success_evidence(self) -> None: + """The real-browser evidence must retain deterministic crash and teardown proof.""" + + runner = RUNNER.read_text(encoding="utf-8") + for expected in ( + '"browser_crash"', + '"browser_process_crash_detected"', + '"browser_process_terminated"', + '"chromium_process_set_terminated"', + "Agent Task browser-crash recovery gate failed", + ): + with self.subTest(expected=expected): + self.assertIn(expected, runner) + + def test_zz_signal_boundary_tests_restore_process_wide_modules(self) -> None: + """Mocked pidfd helpers must not leak process-wide module mutations.""" + + self.assertIs(os.close, ORIGINAL_OS_CLOSE) + self.assertIs(getattr(os, "pidfd_open", None), ORIGINAL_PIDFD_OPEN) + self.assertIs( + getattr(signal, "pidfd_send_signal", None), + ORIGINAL_PIDFD_SEND_SIGNAL, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_chromedriver_error_diagnostic_contract.py b/tests/test_chromedriver_error_diagnostic_contract.py index f9f0f9734..5a5369d6f 100644 --- a/tests/test_chromedriver_error_diagnostic_contract.py +++ b/tests/test_chromedriver_error_diagnostic_contract.py @@ -37,13 +37,21 @@ def log_message(self, _format: str, *args: object) -> None: class ChromeDriverErrorDiagnosticContractTests(unittest.TestCase): """ChromeDriver-controlled response bytes must not be reflected into CI errors.""" - def _request_against(self, *, status: int) -> RuntimeError: + def _request_against( + self, + *, + status: int, + response_body: bytes | None = None, + ) -> RuntimeError: namespace = runpy.run_path(str(RUNNER), run_name="chromedriver_error_diagnostic_contract") json_request = namespace["_json_request"] class Handler(_ErrorResponseHandler): response_status = status + if response_body is not None: + Handler.response_body = response_body + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() @@ -70,6 +78,47 @@ def test_webdriver_error_does_not_reflect_response_message(self) -> None: self.assertIn("WebDriver", str(error)) self.assertNotIn(SECRET_MARKER, str(error)) + def test_session_not_created_retains_only_allowlisted_startup_reason(self) -> None: + """A sandbox startup failure must keep typed safe evidence without raw detail.""" + + sandbox_response = ( + b'{"value":{"error":"session not created","message":"' + b'session not created: Chrome failed to start: No usable sandbox! ' + + SECRET_MARKER.encode("ascii") + + b'"}}' + ) + for status in (500, 200): + with self.subTest(status=status): + error = self._request_against( + status=status, + response_body=sandbox_response, + ) + + self.assertEqual(type(error).__name__, "_WebDriverSessionNotCreatedError") + self.assertEqual(getattr(error, "error_code", None), "session_not_created") + self.assertEqual( + getattr(error, "startup_reason", None), + "sandbox_unavailable", + ) + self.assertNotIn(SECRET_MARKER, str(error)) + self.assertNotIn(SECRET_MARKER, repr(error)) + + def test_unrecognized_session_startup_detail_remains_unknown_and_redacted(self) -> None: + """Unreviewed ChromeDriver prose must not become evidence or a new reason code.""" + + unknown_response = ( + b'{"value":{"error":"session not created","message":"' + + SECRET_MARKER.encode("ascii") + + b'"}}' + ) + error = self._request_against(status=500, response_body=unknown_response) + + self.assertEqual(type(error).__name__, "_WebDriverSessionNotCreatedError") + self.assertEqual(getattr(error, "error_code", None), "session_not_created") + self.assertEqual(getattr(error, "startup_reason", None), "unknown") + self.assertNotIn(SECRET_MARKER, str(error)) + self.assertNotIn(SECRET_MARKER, repr(error)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_chromedriver_process_diagnostic_contract.py b/tests/test_chromedriver_process_diagnostic_contract.py new file mode 100644 index 000000000..aff383553 --- /dev/null +++ b/tests/test_chromedriver_process_diagnostic_contract.py @@ -0,0 +1,197 @@ +"""Contract tests for credential-safe ChromeDriver process-start diagnostics.""" + +from __future__ import annotations + +import io +import os +import pathlib +import runpy +import threading +import time +import unittest +from unittest.mock import patch + + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER_PATH = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class _UnrecognizedStream: + """Simulate a legacy process double with no usable diagnostic bytes.""" + + def read(self, _size: int) -> object: + """Return one deliberately unrecognized value.""" + + return object() + + +class ChromeDriverProcessDiagnosticContractTests(unittest.TestCase): + """Keep process diagnostics bounded to reviewed reason codes and drained continuously.""" + + @classmethod + def setUpClass(cls) -> None: + cls.runner = runpy.run_path( + str(RUNNER_PATH), + run_name="originweave_mv3_process_diagnostic_contract", + ) + + def test_split_sandbox_marker_is_classified_without_raw_retention(self) -> None: + diagnostic_type = self.runner["_ChromeDriverStartupDiagnostic"] + diagnostic = diagnostic_type() + + diagnostic.feed(b"prefix /tmp/private-profile token=do-not-retain No usable sand") + self.assertEqual(diagnostic.startup_reason, "unknown") + diagnostic.feed(b"box! suffix secret-shaped-value") + + self.assertEqual(diagnostic.startup_reason, "sandbox_unavailable") + rendered_state = repr(diagnostic) + self.assertNotIn("private-profile", rendered_state) + self.assertNotIn("do-not-retain", rendered_state) + self.assertNotIn("secret-shaped-value", rendered_state) + self.assertFalse(hasattr(diagnostic, "__dict__")) + + def test_unknown_process_diagnostic_stays_unknown(self) -> None: + diagnostic_type = self.runner["_ChromeDriverStartupDiagnostic"] + diagnostic = diagnostic_type() + + diagnostic.feed(b"session not created: /tmp/private-profile bearer-secret") + + self.assertEqual(diagnostic.startup_reason, "unknown") + self.assertNotIn("private-profile", repr(diagnostic)) + self.assertNotIn("bearer-secret", repr(diagnostic)) + + def test_drain_consumes_large_pipe_without_retaining_payload(self) -> None: + diagnostic_type = self.runner["_ChromeDriverStartupDiagnostic"] + drain = self.runner["_drain_chromedriver_diagnostics"] + diagnostic = diagnostic_type() + read_fd, write_fd = os.pipe() + reader = os.fdopen(read_fd, "rb", buffering=0) + writer = os.fdopen(write_fd, "wb", buffering=0) + thread = threading.Thread(target=drain, args=(reader, diagnostic), daemon=True) + thread.start() + try: + writer.write(b"x" * 262_144) + writer.write(b"No usable sand") + writer.write(b"box!") + finally: + writer.close() + thread.join(timeout=2.0) + reader.close() + + self.assertFalse(thread.is_alive(), "ChromeDriver diagnostic pipe drain deadlocked") + self.assertEqual(diagnostic.startup_reason, "sandbox_unavailable") + self.assertNotIn("x" * 32, repr(diagnostic)) + + def test_bytesio_drain_keeps_unreviewed_text_out_of_state(self) -> None: + diagnostic_type = self.runner["_ChromeDriverStartupDiagnostic"] + drain = self.runner["_drain_chromedriver_diagnostics"] + diagnostic = diagnostic_type() + sensitive = b"/home/runner/private-profile bearer-super-secret" + + drain(io.BytesIO(sensitive), diagnostic) + + self.assertEqual(diagnostic.startup_reason, "unknown") + self.assertNotIn("private-profile", repr(diagnostic)) + self.assertNotIn("super-secret", repr(diagnostic)) + + def test_text_stream_drain_supports_existing_process_doubles_without_retention(self) -> None: + diagnostic_type = self.runner["_ChromeDriverStartupDiagnostic"] + drain = self.runner["_drain_chromedriver_diagnostics"] + diagnostic = diagnostic_type() + sensitive = "prefix /tmp/private-profile No usable sandbox! bearer-super-secret" + + drain(io.StringIO(sensitive), diagnostic) + + self.assertEqual(diagnostic.startup_reason, "sandbox_unavailable") + self.assertNotIn("private-profile", repr(diagnostic)) + self.assertNotIn("super-secret", repr(diagnostic)) + + def test_unrecognized_stream_chunk_is_discarded_without_thread_failure(self) -> None: + """Existing process doubles cannot turn discarded diagnostics into a thread error.""" + + diagnostic_type = self.runner["_ChromeDriverStartupDiagnostic"] + drain = self.runner["_drain_chromedriver_diagnostics"] + diagnostic = diagnostic_type() + + drain(_UnrecognizedStream(), diagnostic) + + self.assertEqual(diagnostic.startup_reason, "unknown") + + def test_session_creation_waits_for_bounded_diagnostic_handoff(self) -> None: + """A delayed reviewed diagnostic refines a concurrent session-creation error.""" + + diagnostic_type = self.runner["_ChromeDriverStartupDiagnostic"] + create_session = self.runner["_create_chromedriver_session"] + session_error_type = self.runner["_WebDriverSessionNotCreatedError"] + diagnostic = diagnostic_type() + session_started = threading.Event() + + def delayed_diagnostic() -> None: + session_started.wait(timeout=1.0) + time.sleep(0.01) + diagnostic.feed(b"No usable sandbox") + + def rejected_session(*_args: object, **_kwargs: object) -> dict[str, object]: + session_started.set() + raise session_error_type("unknown") + + feeder = threading.Thread(target=delayed_diagnostic) + feeder.start() + with patch.dict( + create_session.__globals__, + {"_wait_for_driver": lambda _port: None, "_json_request": rejected_session}, + ), self.assertRaises(session_error_type) as captured: + create_session(9515, {}, diagnostic) + feeder.join(timeout=1.0) + + self.assertFalse(feeder.is_alive()) + self.assertEqual(captured.exception.startup_reason, "sandbox_unavailable") + + def test_session_creation_waits_past_unreviewed_output_for_delayed_marker(self) -> None: + """An earlier verbose chunk cannot end the reviewed-reason handoff.""" + + diagnostic_type = self.runner["_ChromeDriverStartupDiagnostic"] + create_session = self.runner["_create_chromedriver_session"] + session_error_type = self.runner["_WebDriverSessionNotCreatedError"] + diagnostic = diagnostic_type() + diagnostic.feed(b"ordinary ChromeDriver startup line") + session_started = threading.Event() + + def delayed_marker() -> None: + session_started.wait(timeout=1.0) + time.sleep(0.01) + diagnostic.feed(b"No usable sandbox") + + def rejected_session(*_args: object, **_kwargs: object) -> dict[str, object]: + session_started.set() + raise session_error_type("unknown") + + feeder = threading.Thread(target=delayed_marker) + feeder.start() + with patch.dict( + create_session.__globals__, + {"_wait_for_driver": lambda _port: None, "_json_request": rejected_session}, + ), self.assertRaises(session_error_type) as captured: + create_session(9515, {}, diagnostic) + feeder.join(timeout=1.0) + + self.assertFalse(feeder.is_alive()) + self.assertEqual(captured.exception.startup_reason, "sandbox_unavailable") + + def test_all_chromedriver_launches_stream_instead_of_discarding_output(self) -> None: + source = RUNNER_PATH.read_text(encoding="utf-8") + + self.assertNotIn("stdout=subprocess.DEVNULL", source) + self.assertIn("stdout=subprocess.PIPE", source) + self.assertEqual(source.count("_start_chromedriver("), 5) + self.assertEqual(source.count("_create_chromedriver_session("), 5) + + def test_shared_launch_enables_verbose_diagnostics_without_log_file(self) -> None: + source = RUNNER_PATH.read_text(encoding="utf-8") + + self.assertIn('"--verbose"', source) + self.assertNotIn("--log-path", source) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mv3_browser_sandbox_contract.py b/tests/test_mv3_browser_sandbox_contract.py new file mode 100644 index 000000000..c77bd0a52 --- /dev/null +++ b/tests/test_mv3_browser_sandbox_contract.py @@ -0,0 +1,33 @@ +"""Regression contract for preserving Chromium sandboxing in real-browser evidence.""" + +from __future__ import annotations + +import inspect +import pathlib +import runpy +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class BrowserSandboxContractTests(unittest.TestCase): + """Require every real-Chromium evidence path to keep process sandboxing enabled.""" + + def test_real_browser_passes_do_not_disable_chromium_sandbox(self) -> None: + """No real-browser evidence lane may launch Chrome with ``--no-sandbox``.""" + + namespace = runpy.run_path(str(RUNNER), run_name="browser_sandbox_contract") + for function_name in ( + "_run_browser_pass", + "_run_agent_task_browser_pass", + "_run_agent_task_forced_close_browser_pass", + "_run_agent_task_browser_crash_browser_pass", + ): + with self.subTest(function_name=function_name): + source = inspect.getsource(namespace[function_name]) + self.assertNotIn('"--no-sandbox"', source) + + +if __name__ == "__main__": + unittest.main()