test(browser): locate controlled Agent Task targets semantically

scripts/ci/run_mv3_compatibility.py

@@ -7,15 +7,17 @@
 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
 
+import contextlib
+import hashlib
 import http.client
 import http.server
 import json
@@ -45,6 +47,8 @@
 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_AGENT_TASK_STRUCTURED_VALUE_BYTES = 4_096
 MAX_U64 = (1 << 64) - 1
 W3C_ELEMENT_KEY = "element-6066-11e4-a52e-4f735466cecf"
 PATH_TOKEN_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_.")
@@ -239,6 +243,60 @@ 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 _hash_agent_task_structured_value(value: str) -> str:
+    """Hash one bounded extracted text value without retaining the raw value in evidence."""
+
+    if not isinstance(value, str):
+        raise TypeError("Agent Task structured value must be text")
+    encoded = value.encode("utf-8")
+    if not encoded:
+        raise ValueError("Agent Task structured value must not be empty")
+    if len(encoded) > MAX_AGENT_TASK_STRUCTURED_VALUE_BYTES:
+        raise ValueError("Agent Task structured value exceeded the bounded text contract")
+    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+
+
 def _parse_linux_proc_status_rss_bytes(status_text: str) -> int:
     """Parse exactly one positive Linux ``VmRSS`` kB field into bounded bytes."""
 
@@ -833,18 +891,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,
@@ -900,7 +964,19 @@ def _run_agent_task_browser_pass(
         if not url_unchanged:
             raise RuntimeError("Agent Task URL changed during submission")
 
-        result_element = _find_element(driver_port, session_id, "#task-result")
+        result_element = _find_element_by_accessible_role_name(
+            driver_port,
+            session_id,
+            "status",
+            "Task result",
+        )
+        result_role, result_name = _get_element_semantics(
+            driver_port,
+            session_id,
+            result_element,
+        )
+        if result_role != "status" or result_name != "Task result":
+            raise RuntimeError("Agent Task result semantic evidence mismatch")
         state = _json_request(
             driver_port,
             "GET",
@@ -914,6 +990,7 @@ def _run_agent_task_browser_pass(
         _validate_agent_task_submitted_state(state)
         if text != AGENT_TASK_INPUT_VALUE:
             raise RuntimeError("Agent Task result did not match the synthetic typed value")
+        structured_value_sha256 = _hash_agent_task_structured_value(text)
         process_evidence = _snapshot_linux_process_evidence()
         chromium_process_ids = _discover_linux_process_tree_ids(
             browser_process_id,
@@ -937,6 +1014,9 @@ def _run_agent_task_browser_pass(
             "url_unchanged": url_unchanged,
             "input_semantics_verified": True,
             "submit_semantics_verified": True,
+            "result_semantics_verified": True,
+            "structured_value_field": "task_result",
+            "structured_value_sha256": structured_value_sha256,
             "extensions_disabled": True,
             "browser_process_rss_bytes": browser_process_rss_bytes,
             "chromium_process_count": len(chromium_process_ids),
@@ -1011,6 +1091,9 @@ def _run_agent_task_trial(
         "url_unchanged": result["url_unchanged"],
         "input_semantics_verified": result["input_semantics_verified"],
         "submit_semantics_verified": result["submit_semantics_verified"],
+        "result_semantics_verified": result["result_semantics_verified"],
+        "structured_value_field": result["structured_value_field"],
+        "structured_value_sha256": result["structured_value_sha256"],
         "extensions_disabled": result["extensions_disabled"],
         "browser_process_rss_bytes": result["browser_process_rss_bytes"],
         "chromium_process_count": result["chromium_process_count"],
@@ -1035,6 +1118,11 @@ def _agent_task_surfaces_complete(agent_task_trials: list[dict[str, Any]]) -> bo
         and trial.get("url_unchanged") is True
         and trial.get("input_semantics_verified") is True
         and trial.get("submit_semantics_verified") is True
+        and trial.get("result_semantics_verified") is True
+        and trial.get("structured_value_field") == "task_result"
+        and isinstance(trial.get("structured_value_sha256"), str)
+        and len(trial["structured_value_sha256"]) == len("sha256:") + 64
+        and trial["structured_value_sha256"].startswith("sha256:")
         and trial.get("extensions_disabled") is True
         and trial.get("profile_cleaned") is True
         and isinstance(trial.get("browser_process_rss_bytes"), int)

tests/fixtures/agent_task_basic/index.html

@@ -16,7 +16,12 @@ <h1>Controlled Agent Task</h1>
       <button type="submit">Submit task</button>
     </form>
 
-    <output id="task-result" data-state="idle" aria-live="polite">idle</output>
+    <output
+      id="task-result"
+      data-state="idle"
+      aria-label="Task result"
+      aria-live="polite"
+    >idle</output>
 
     <p
       data-originweave-untrusted="prompt-injection"

tests/test_agent_task_pinned_chrome_contract.py

@@ -210,6 +210,9 @@ def test_agent_task_surface_completeness_is_non_vacuous(self) -> None:
                         "url_unchanged": True,
                         "input_semantics_verified": True,
                         "submit_semantics_verified": True,
+                        "result_semantics_verified": True,
+                        "structured_value_field": "task_result",
+                        "structured_value_sha256": "sha256:" + "0" * 64,
                         "extensions_disabled": True,
                         "profile_cleaned": True,
                         "browser_process_rss_bytes": 1,
@@ -317,6 +320,9 @@ def successful_agent_task_trial(*_args: object, **_kwargs: object) -> dict[str,
                 "url_unchanged": True,
                 "input_semantics_verified": True,
                 "submit_semantics_verified": True,
+                "result_semantics_verified": True,
+                "structured_value_field": "task_result",
+                "structured_value_sha256": "sha256:" + "0" * 64,
                 "extensions_disabled": True,
                 "browser_process_rss_bytes": 1,
                 "chromium_process_count": 1,

tests/test_agent_task_structured_value_contract.py

@@ -0,0 +1,66 @@
+"""Contract for credential-safe structured extraction in the controlled Agent Task."""
+
+from __future__ import annotations
+
+import hashlib
+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"
+
+
+class AgentTaskStructuredValueContractTests(unittest.TestCase):
+    """Require bounded semantic extraction without retaining the extracted value."""
+
+    def test_result_is_discovered_by_exact_browser_semantics(self) -> None:
+        """The result node must be located by browser-computed role/name, not fixture CSS."""
+
+        runner = RUNNER.read_text(encoding="utf-8")
+        fixture = FIXTURE.read_text(encoding="utf-8")
+        self.assertIn('aria-label="Task result"', fixture)
+        self.assertIn('"status"', runner)
+        self.assertIn('"Task result"', runner)
+        self.assertIn('"result_semantics_verified"', runner)
+        self.assertNotIn('_find_element(driver_port, session_id, "#task-result")', runner)
+
+    def test_structured_value_hash_is_bounded_and_canonical(self) -> None:
+        """Only a canonical SHA-256 digest may leave the controlled extraction boundary."""
+
+        namespace = runpy.run_path(str(RUNNER), run_name="agent_task_structured_value_contract")
+        self.assertIn("MAX_AGENT_TASK_STRUCTURED_VALUE_BYTES", namespace)
+        self.assertIn("_hash_agent_task_structured_value", namespace)
+        helper = namespace["_hash_agent_task_structured_value"]
+        maximum = namespace["MAX_AGENT_TASK_STRUCTURED_VALUE_BYTES"]
+
+        value = "synthetic structured value"
+        expected = "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()
+        digest = helper(value)
+        self.assertEqual(digest, expected)
+        self.assertNotIn(value, digest)
+        self.assertEqual(len(digest), len("sha256:") + 64)
+
+        with self.assertRaises(ValueError):
+            helper("")
+        with self.assertRaises(ValueError):
+            helper("x" * (maximum + 1))
+        with self.assertRaises(TypeError):
+            helper(42)
+
+    def test_agent_task_evidence_reports_field_and_digest_not_raw_result(self) -> None:
+        """Trial evidence must expose a field identifier and digest, not extracted text."""
+
+        runner = RUNNER.read_text(encoding="utf-8")
+        for expected in (
+            '"structured_value_field"',
+            '"structured_value_sha256"',
+            '"task_result"',
+        ):
+            with self.subTest(expected=expected):
+                self.assertIn(expected, runner)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mv3_compatibility_contract.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import http.client
 import json
 import pathlib
 import runpy
+import tempfile
 import unittest
 
 ROOT = pathlib.Path(__file__).resolve().parents[1]
@@ -120,6 +122,32 @@ def test_runner_transport_cannot_follow_dynamic_url_schemes(self) -> None:
         self.assertNotIn("urllib.request", runner)
         self.assertNotIn("urllib.error", runner)
 
+    def test_fixture_handler_cannot_follow_symlinks_outside_its_root(self) -> None:
+        """The loopback fixture server must not expose files outside its configured root."""
+
+        namespace = runpy.run_path(str(RUNNER), run_name="mv3_contract")
+        start_server = namespace["_start_fixture_server"]
+        stop_server = namespace["_stop_fixture_server"]
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            temporary_root = pathlib.Path(temporary_directory)
+            fixture_root = temporary_root / "fixture"
+            outside_root = temporary_root / "outside"
+            fixture_root.mkdir()
+            outside_root.mkdir()
+            (outside_root / "secret.txt").write_text("outside", encoding="utf-8")
+            (fixture_root / "escape").symlink_to(outside_root, target_is_directory=True)
+
+            server, thread = start_server(fixture_root)
+            connection = http.client.HTTPConnection(*server.server_address, timeout=1)
+            try:
+                connection.request("GET", "/escape/secret.txt")
+                response = connection.getresponse()
+                self.assertEqual(response.status, 404)
+                self.assertNotIn(b"outside", response.read())
+            finally:
+                connection.close()
+                stop_server(server, thread)
+
     def test_runner_accepts_real_chromedriver_element_ids_without_path_injection(self) -> None:
         """ChromeDriver dotted element IDs must work while path syntax stays fail-closed."""