diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index 28a3fb1e..131edfd0 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -1,12 +1,15 @@ #!/usr/bin/env python3 -"""Run bounded repeatable Manifest V3 compatibility evidence against pinned Chromium. +"""Run bounded repeatable real-browser evidence against pinned Chromium. This is a release/CI evidence runner, not a product browser adapter. It uses the 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 -restart-persistence behavior. +restart-persistence behavior. It also executes the controlled Agent Task fixture +with extensions disabled in a fresh profile, performs real WebDriver input and +click operations, verifies the observable post-condition, and proves profile +cleanup without treating page content as instruction or authority. """ from __future__ import annotations @@ -27,9 +30,12 @@ ROOT = pathlib.Path(__file__).resolve().parents[2] FIXTURE = ROOT / "tests" / "fixtures" / "mv3_basic" +AGENT_TASK_FIXTURE = ROOT / "tests/fixtures/agent_task_basic" PINNED_CHROME_VERSION = "150.0.7871.129" PINNED_CHROME_REVISION = "r1639810" REPEATABILITY_TRIALS = 3 +AGENT_TASK_REPEATABILITY_TRIALS = 3 +AGENT_TASK_INPUT_VALUE = "originweave controlled input" REQUEST_TIMEOUT_SECONDS = 5.0 STARTUP_TIMEOUT_SECONDS = 20.0 FIXTURE_TIMEOUT_SECONDS = 20.0 @@ -150,6 +156,29 @@ def _execute(driver_port: int, session_id: str, script: str) -> Any: return response.get("value") +def _find_element(driver_port: int, session_id: str, selector: str) -> str: + """Find one fixture element and return its validated ChromeDriver identifier.""" + + found = _json_request( + driver_port, + "POST", + _webdriver_path(session_id, "/element"), + {"using": "css selector", "value": selector}, + ) + element = found.get("value", {}) + element_id = element.get(W3C_ELEMENT_KEY) if isinstance(element, dict) else None + if not isinstance(element_id, str): + raise RuntimeError("WebDriver did not return a W3C element identifier") + return _path_token(element_id, "element identifier") + + +def _element_command_path(session_id: str, element_id: str, suffix: str) -> str: + """Build a bounded WebDriver element command path from validated identifiers.""" + + safe_element = _path_token(element_id, "element identifier") + return _webdriver_path(session_id, f"/element/{safe_element}{suffix}") + + def _wait_for_extension_evidence( driver_port: int, session_id: str, @@ -220,37 +249,18 @@ def _wait_for_extension_evidence( def _exercise_real_click(driver_port: int, session_id: str) -> str: """Use the WebDriver element-click command and verify the DOM post-condition.""" - found = _json_request( - driver_port, - "POST", - _webdriver_path(session_id, "/element"), - {"using": "css selector", "value": "#fixture-button"}, - ) - element = found.get("value", {}) - element_id = element.get(W3C_ELEMENT_KEY) if isinstance(element, dict) else None - if not isinstance(element_id, str): - raise RuntimeError("WebDriver did not return a W3C element identifier") - safe_element = _path_token(element_id, "element identifier") + safe_element = _find_element(driver_port, session_id, "#fixture-button") _json_request( driver_port, "POST", - _webdriver_path(session_id, f"/element/{safe_element}/click"), + _element_command_path(session_id, safe_element, "/click"), {}, ) - output = _json_request( - driver_port, - "POST", - _webdriver_path(session_id, "/element"), - {"using": "css selector", "value": "#fixture-output"}, - ).get("value", {}) - output_id = output.get(W3C_ELEMENT_KEY) if isinstance(output, dict) else None - if not isinstance(output_id, str): - raise RuntimeError("WebDriver did not return the fixture output element") - safe_output = _path_token(output_id, "element identifier") + safe_output = _find_element(driver_port, session_id, "#fixture-output") text = _json_request( driver_port, "GET", - _webdriver_path(session_id, f"/element/{safe_output}/text"), + _element_command_path(session_id, safe_output, "/text"), ).get("value") if text != "clicked": raise RuntimeError(f"real click post-condition failed: {text!r}") @@ -433,8 +443,220 @@ def _run_restart_trial( } +def _run_agent_task_browser_pass( + chrome_bin: pathlib.Path, + chromedriver_bin: pathlib.Path, + fixture_url: str, + profile_dir: str, +) -> dict[str, Any]: + """Execute one synthetic Agent Task through real WebDriver input in pinned Chrome.""" + + started = time.monotonic() + driver_port = _free_loopback_port() + session_id: str | 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, + ) + try: + _wait_for_driver(driver_port) + session = _json_request( + driver_port, + "POST", + "/session", + { + "capabilities": { + "alwaysMatch": { + "browserName": "chrome", + "goog:chromeOptions": { + "binary": str(chrome_bin), + "args": [ + "--headless=new", + "--no-first-run", + "--disable-default-apps", + "--disable-component-update", + "--disable-sync", + "--disable-dev-shm-usage", + "--no-sandbox", + "--disable-extensions", + f"--user-data-dir={profile_dir}", + ], + }, + } + } + }, + ).get("value", {}) + if not isinstance(session, dict): + raise RuntimeError("ChromeDriver Agent Task 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 an Agent Task session id") + session_id = _path_token(raw_session_id, "session identifier") + browser_version = ( + capabilities.get("browserVersion") if isinstance(capabilities, dict) else None + ) + if browser_version != PINNED_CHROME_VERSION: + raise RuntimeError( + f"unexpected Agent Task Chrome version: expected {PINNED_CHROME_VERSION}, " + f"got {browser_version!r}" + ) + + _json_request( + driver_port, + "POST", + _webdriver_path(session_id, "/url"), + {"url": fixture_url}, + ) + initial_url = _json_request( + driver_port, + "GET", + _webdriver_path(session_id, "/url"), + ).get("value") + if initial_url != fixture_url: + raise RuntimeError( + f"Agent Task initial URL mismatch: expected {fixture_url!r}, got {initial_url!r}" + ) + input_element = _find_element(driver_port, session_id, "#task-text") + _json_request( + driver_port, + "POST", + _element_command_path(session_id, input_element, "/clear"), + {}, + ) + _json_request( + driver_port, + "POST", + _element_command_path(session_id, input_element, "/value"), + {"text": AGENT_TASK_INPUT_VALUE, "value": list(AGENT_TASK_INPUT_VALUE)}, + ) + submit_element = _find_element( + driver_port, + session_id, + "#agent-task-form button[type=submit]", + ) + _json_request( + driver_port, + "POST", + _element_command_path(session_id, submit_element, "/click"), + {}, + ) + post_submit_url = _json_request( + driver_port, + "GET", + _webdriver_path(session_id, "/url"), + ).get("value") + url_unchanged = post_submit_url == initial_url + if not url_unchanged: + raise RuntimeError("Agent Task URL changed during submission") + result_element = _find_element(driver_port, session_id, "#task-result") + state = _json_request( + driver_port, + "GET", + _element_command_path(session_id, result_element, "/attribute/data-state"), + ).get("value") + text = _json_request( + driver_port, + "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}") + if text != AGENT_TASK_INPUT_VALUE: + raise RuntimeError("Agent Task result did not match the synthetic typed value") + return { + "browser_version": browser_version, + "post_condition": True, + "input_echo_verified": True, + "url_unchanged": url_unchanged, + "extensions_disabled": True, + "duration_ms": round((time.monotonic() - started) * 1000), + } + 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) + + +def _run_agent_task_trial( + chrome_bin: pathlib.Path, + chromedriver_bin: pathlib.Path, + fixture_url: str, + trial_number: int, +) -> dict[str, Any]: + """Run one isolated Agent Task browser trial and prove its profile is removed.""" + + trial_started = time.monotonic() + profile_path: pathlib.Path + with tempfile.TemporaryDirectory( + prefix=f"originweave-agent-task-trial-{trial_number}-" + ) as profile_dir: + profile_path = pathlib.Path(profile_dir) + result = _run_agent_task_browser_pass( + chrome_bin, + chromedriver_bin, + fixture_url, + profile_dir, + ) + profile_cleaned = not profile_path.exists() + if not profile_cleaned: + raise RuntimeError(f"Agent Task profile cleanup failed in trial {trial_number}") + + return { + "trial_number": trial_number, + "passed": True, + "browser_version": result["browser_version"], + "post_condition": result["post_condition"], + "input_echo_verified": result["input_echo_verified"], + "url_unchanged": result["url_unchanged"], + "extensions_disabled": result["extensions_disabled"], + "profile_cleaned": profile_cleaned, + "duration_ms": round((time.monotonic() - trial_started) * 1000), + } + + +def _start_fixture_server(directory: pathlib.Path) -> tuple[http.server.ThreadingHTTPServer, threading.Thread]: + """Start one loopback-only static fixture server for a bounded browser lane.""" + + server = http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), + lambda *args, **kwargs: QuietFixtureHandler( + *args, + directory=str(directory), + **kwargs, + ), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, thread + + +def _stop_fixture_server( + server: http.server.ThreadingHTTPServer, + thread: threading.Thread, +) -> None: + """Stop one bounded fixture server and join its helper thread.""" + + server.shutdown() + server.server_close() + thread.join(timeout=5) + + def main() -> int: - """Run three independent restart trials and emit bounded repeatability evidence.""" + """Run bounded MV3 and Agent Task trials and emit credential-free evidence.""" chrome_bin = pathlib.Path(os.environ.get("CHROME_BIN", "")) chromedriver_bin = pathlib.Path(os.environ.get("CHROMEDRIVER_BIN", "")) @@ -444,15 +666,11 @@ def main() -> int: raise SystemExit("CHROMEDRIVER_BIN must point to the matching pinned ChromeDriver") if not (FIXTURE / "manifest.json").is_file(): raise SystemExit("MV3 fixture manifest is missing") + if not (AGENT_TASK_FIXTURE / "index.html").is_file(): + raise SystemExit("Agent Task fixture is missing") - fixture_server = http.server.ThreadingHTTPServer( - ("127.0.0.1", 0), - lambda *args, **kwargs: QuietFixtureHandler( - *args, directory=str(FIXTURE), **kwargs - ), - ) - fixture_thread = threading.Thread(target=fixture_server.serve_forever, daemon=True) - fixture_thread.start() + fixture_server, fixture_thread = _start_fixture_server(FIXTURE) + agent_task_server, agent_task_thread = _start_fixture_server(AGENT_TASK_FIXTURE) started = time.monotonic() try: @@ -496,6 +714,44 @@ def main() -> int: for name in first_surfaces } + agent_task_url = ( + f"http://127.0.0.1:{agent_task_server.server_port}/index.html" + ) + agent_task_trials: list[dict[str, Any]] = [] + for trial_number in range(1, AGENT_TASK_REPEATABILITY_TRIALS + 1): + try: + agent_task_trials.append( + _run_agent_task_trial( + chrome_bin, + chromedriver_bin, + agent_task_url, + trial_number, + ) + ) + except (OSError, ValueError, RuntimeError, json.JSONDecodeError): + agent_task_trials.append( + { + "trial_number": trial_number, + "passed": False, + } + ) + + agent_task_successful_trials = sum( + 1 for trial in agent_task_trials if trial.get("passed") is True + ) + agent_task_trial_pass_rate = ( + agent_task_successful_trials / AGENT_TASK_REPEATABILITY_TRIALS + ) + agent_task_surfaces_complete = all( + trial.get("post_condition") is True + and trial.get("input_echo_verified") is True + and trial.get("url_unchanged") is True + and trial.get("extensions_disabled") is True + and trial.get("profile_cleaned") is True + for trial in agent_task_trials + if trial.get("passed") is True + ) + evidence = { "chrome_version": PINNED_CHROME_VERSION, "chrome_revision": PINNED_CHROME_REVISION, @@ -509,6 +765,12 @@ def main() -> int: if successful_results else [] ), + "agent_task": { + "repeatability_trials": AGENT_TASK_REPEATABILITY_TRIALS, + "successful_trials": agent_task_successful_trials, + "trial_pass_rate": agent_task_trial_pass_rate, + "trial_results": agent_task_trials, + }, "duration_ms": round((time.monotonic() - started) * 1000), } print(json.dumps(evidence, sort_keys=True)) @@ -519,11 +781,18 @@ def main() -> int: ) if not common_surfaces or not all(common_surfaces.values()): raise RuntimeError("Manifest V3 repeatability surfaces were incomplete") + if agent_task_successful_trials != AGENT_TASK_REPEATABILITY_TRIALS: + raise RuntimeError( + "Agent Task repeatability gate failed: " + f"{agent_task_successful_trials}/{AGENT_TASK_REPEATABILITY_TRIALS} " + "trials passed" + ) + if not agent_task_surfaces_complete: + raise RuntimeError("Agent Task repeatability surfaces were incomplete") return 0 finally: - fixture_server.shutdown() - fixture_server.server_close() - fixture_thread.join(timeout=5) + _stop_fixture_server(agent_task_server, agent_task_thread) + _stop_fixture_server(fixture_server, fixture_thread) if __name__ == "__main__": diff --git a/tests/test_agent_task_pinned_chrome_contract.py b/tests/test_agent_task_pinned_chrome_contract.py new file mode 100644 index 00000000..bf084090 --- /dev/null +++ b/tests/test_agent_task_pinned_chrome_contract.py @@ -0,0 +1,77 @@ +"""Contract for executing the controlled Agent Task fixture on pinned Chrome.""" + +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" +FIXTURE = ROOT / "tests" / "fixtures" / "agent_task_basic" / "index.html" +WORKFLOW = ROOT / ".github" / "workflows" / "mv3-compatibility.yml" + + +class AgentTaskPinnedChromeContractTests(unittest.TestCase): + """Keep the first real-browser Agent Task evidence bounded and reproducible.""" + + def test_runner_exposes_a_separate_agent_task_browser_boundary(self) -> None: + """The pinned-browser runner must execute the controlled Agent Task fixture.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_contract") + for expected in ( + "AGENT_TASK_FIXTURE", + "AGENT_TASK_REPEATABILITY_TRIALS", + "_run_agent_task_browser_pass", + "_run_agent_task_trial", + ): + with self.subTest(expected=expected): + self.assertIn(expected, namespace) + + def test_agent_task_pass_uses_real_webdriver_input_and_post_condition(self) -> None: + """The evidence lane must type, click, and verify fixture state in real Chrome.""" + + runner = RUNNER.read_text(encoding="utf-8") + for expected in ( + "tests/fixtures/agent_task_basic", + '"--disable-extensions"', + '"/value"', + '"/click"', + '"/attribute/data-state"', + '"submitted"', + '"profile_cleaned"', + '"agent_task"', + '"trial_pass_rate"', + ): + with self.subTest(expected=expected): + self.assertIn(expected, runner) + + def test_agent_task_submission_preserves_the_loaded_url(self) -> None: + """Submission must prove that the controlled action did not navigate away.""" + + runner = RUNNER.read_text(encoding="utf-8") + for expected in ( + "initial_url", + "post_submit_url", + "url_unchanged", + ): + with self.subTest(expected=expected): + self.assertIn(expected, runner) + self.assertIn("Agent Task URL changed during submission", runner) + + def test_agent_task_fixture_runs_under_the_existing_pinned_chrome_job(self) -> None: + """No floating browser or second workflow may be introduced for this slice.""" + + workflow = WORKFLOW.read_text(encoding="utf-8") + runner = RUNNER.read_text(encoding="utf-8") + self.assertTrue(FIXTURE.is_file()) + self.assertIn('CHROME_VERSION: "150.0.7871.129"', workflow) + self.assertIn("run_mv3_compatibility.py", workflow) + self.assertIn("150.0.7871.129", runner) + self.assertNotIn("google-chrome-stable", runner.lower()) + self.assertNotIn("COPILOT_GITHUB_TOKEN", runner) + self.assertNotIn("NVIDIA_NIM_API_KEY", runner) + + +if __name__ == "__main__": + unittest.main()