test(browser): locate controlled Agent Task targets semantically

scripts/ci/run_mv3_compatibility.py

@@ -7,11 +7,11 @@
 content-script, storage, declarative-net-request, tabs, windows, scripting,
 commands, side-panel, bookmarks, history, real browser-click, and
 restart-persistence behavior. It also executes the controlled Agent Task fixture
-with extensions disabled in a fresh profile, verifies browser-computed role/name
-for the controlled action targets, performs real WebDriver input and click
-operations, verifies the observable post-condition, proves the controlled action
-preserves its loaded URL, and records bounded runtime resource evidence without
-treating page content as instruction or authority.
+with extensions disabled in a fresh profile, locates the controlled action
+targets by exact browser-computed role/name evidence, performs real WebDriver
+input and click operations, verifies the observable post-condition, proves the
+controlled action preserves its loaded URL, and records bounded runtime resource
+evidence without treating page content as instruction or authority.
 """
 
 from __future__ import annotations
@@ -45,6 +45,7 @@
 MAX_PROC_STATUS_CHARACTERS = 65_536
 MAX_BROWSER_PROCESS_TREE_SIZE = 256
 MAX_PROC_PROCESS_SCAN_SIZE = 32_768
+MAX_SEMANTIC_LOCATOR_CANDIDATES = 128
 MAX_U64 = (1 << 64) - 1
 W3C_ELEMENT_KEY = "element-6066-11e4-a52e-4f735466cecf"
 PATH_TOKEN_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_.")
@@ -207,6 +208,47 @@ def _get_element_semantics(
     return role, label
 
 
+def _find_element_by_accessible_role_name(
+    driver_port: int,
+    session_id: str,
+    role: str,
+    accessible_name: str,
+) -> str:
+    """Find exactly one controlled element by browser-computed role and name."""
+
+    found = _json_request(
+        driver_port,
+        "POST",
+        _webdriver_path(session_id, "/elements"),
+        {"using": "css selector", "value": "*"},
+    )
+    elements = found.get("value")
+    if not isinstance(elements, list):
+        raise RuntimeError("WebDriver did not return a semantic locator candidate list")
+    if len(elements) > MAX_SEMANTIC_LOCATOR_CANDIDATES:
+        raise RuntimeError("semantic locator exceeded bounded candidate limit")
+
+    matches: list[str] = []
+    for element in elements:
+        element_id = element.get(W3C_ELEMENT_KEY) if isinstance(element, dict) else None
+        if not isinstance(element_id, str):
+            raise RuntimeError("WebDriver returned malformed semantic locator candidate")
+        safe_element = _path_token(element_id, "element identifier")
+        candidate_role, candidate_name = _get_element_semantics(
+            driver_port,
+            session_id,
+            safe_element,
+        )
+        if candidate_role == role and candidate_name == accessible_name:
+            matches.append(safe_element)
+            if len(matches) > 1:
+                raise RuntimeError("semantic locator returned multiple exact matches")
+
+    if not matches:
+        raise RuntimeError("semantic locator returned no exact match")
+    return matches[0]
+
+
 def _parse_linux_proc_status_rss_bytes(status_text: str) -> int:
     """Parse exactly one positive Linux ``VmRSS`` kB field into bounded bytes."""
 
@@ -395,6 +437,17 @@ def _sample_linux_process_set_rss_bytes(
     return total_rss_bytes
 
 
+def _sample_linux_process_snapshot_rss_bytes(
+    process_id: int,
+    process_evidence: dict[int, tuple[int, int | None]],
+) -> int:
+    """Sample one process RSS from the same bounded snapshot as its process set."""
+
+    if process_id not in process_evidence:
+        raise RuntimeError("Linux process snapshot did not contain the browser root PID")
+    return _sample_linux_process_set_rss_bytes((process_id,), process_evidence)
+
+
 def _wait_for_extension_evidence(
     driver_port: int,
     session_id: str,
@@ -748,18 +801,24 @@ def _run_agent_task_browser_pass(
         if initial_url != fixture_url:
             raise RuntimeError("Agent Task did not load the requested fixture URL")
 
-        input_element = _find_element(driver_port, session_id, "#task-text")
+        input_element = _find_element_by_accessible_role_name(
+            driver_port,
+            session_id,
+            "textbox",
+            "Task text",
+        )
         input_role, input_name = _get_element_semantics(
             driver_port,
             session_id,
             input_element,
         )
         if input_role != "textbox" or input_name != "Task text":
             raise RuntimeError("Agent Task input semantic evidence mismatch")
-        submit_element = _find_element(
+        submit_element = _find_element_by_accessible_role_name(
             driver_port,
             session_id,
-            "#agent-task-form button[type=submit]",
+            "button",
+            "Submit task",
         )
         submit_role, submit_name = _get_element_semantics(
             driver_port,
@@ -835,7 +894,10 @@ def _run_agent_task_browser_pass(
             browser_process_id,
             process_evidence,
         )
-        browser_process_rss_bytes = _sample_linux_process_rss_bytes(browser_process_id)
+        browser_process_rss_bytes = _sample_linux_process_snapshot_rss_bytes(
+            browser_process_id,
+            process_evidence,
+        )
         chromium_process_set_rss_bytes = _sample_linux_process_set_rss_bytes(
             chromium_process_ids,
             process_evidence,

tests/test_agent_task_pinned_chrome_contract.py

@@ -93,6 +93,80 @@ def test_agent_task_observes_computed_role_and_name_before_action(self) -> None:
             with self.subTest(expected=expected):
                 self.assertIn(expected, runner)
 
+    def test_agent_task_locates_controlled_targets_by_exact_role_and_name(self) -> None:
+        """The controlled task must discover targets semantically rather than by fixture CSS."""
+
+        namespace = runpy.run_path(str(RUNNER), run_name="agent_task_semantic_locator")
+        runner = RUNNER.read_text(encoding="utf-8")
+        self.assertIn("_find_element_by_accessible_role_name", namespace)
+        for expected in (
+            "MAX_SEMANTIC_LOCATOR_CANDIDATES",
+            '"/elements"',
+            '"css selector"',
+            '"*"',
+            '"semantic locator returned no exact match"',
+            '"semantic locator returned multiple exact matches"',
+        ):
+            with self.subTest(expected=expected):
+                self.assertIn(expected, runner)
+        self.assertNotIn('_find_element(driver_port, session_id, "#task-text")', runner)
+        self.assertNotIn('_find_element(driver_port, session_id, "#submit-task")', runner)
+
+    def test_semantic_role_name_locator_fails_closed_on_ambiguous_candidates(self) -> None:
+        """Exact semantic discovery must reject zero, duplicate, malformed, and oversized sets."""
+
+        namespace = runpy.run_path(str(RUNNER), run_name="agent_task_locator_behavior")
+        locate = namespace["_find_element_by_accessible_role_name"]
+        element_key = namespace["W3C_ELEMENT_KEY"]
+        candidate_limit = namespace["MAX_SEMANTIC_LOCATOR_CANDIDATES"]
+
+        def install_candidates(
+            candidate_ids: list[str],
+            semantics: dict[str, tuple[str, str]],
+        ) -> None:
+            locate.__globals__["_json_request"] = lambda *_args, **_kwargs: {
+                "value": [{element_key: candidate_id} for candidate_id in candidate_ids]
+            }
+            locate.__globals__["_get_element_semantics"] = (
+                lambda _port, _session, candidate_id: semantics[candidate_id]
+            )
+
+        install_candidates(
+            ["candidate-a", "candidate-b"],
+            {
+                "candidate-a": ("button", "Other"),
+                "candidate-b": ("button", "Submit task"),
+            },
+        )
+        self.assertEqual(locate(4444, "session-a", "button", "Submit task"), "candidate-b")
+
+        install_candidates(["candidate-a"], {"candidate-a": ("button", "Other")})
+        with self.assertRaisesRegex(RuntimeError, "no exact match"):
+            locate(4444, "session-a", "button", "Submit task")
+
+        install_candidates(
+            ["candidate-a", "candidate-b"],
+            {
+                "candidate-a": ("button", "Submit task"),
+                "candidate-b": ("button", "Submit task"),
+            },
+        )
+        with self.assertRaisesRegex(RuntimeError, "multiple exact matches"):
+            locate(4444, "session-a", "button", "Submit task")
+
+        install_candidates(
+            [f"candidate-{index}" for index in range(candidate_limit + 1)],
+            {},
+        )
+        with self.assertRaisesRegex(RuntimeError, "bounded candidate limit"):
+            locate(4444, "session-a", "button", "Submit task")
+
+        locate.__globals__["_json_request"] = lambda *_args, **_kwargs: {
+            "value": [{"not-an-element-id": "candidate-a"}]
+        }
+        with self.assertRaisesRegex(RuntimeError, "malformed semantic locator candidate"):
+            locate(4444, "session-a", "button", "Submit task")
+
     def test_agent_task_records_real_bounded_resource_evidence(self) -> None:
         """The real task must report measured browser/runtime resource evidence."""