diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index e5be0e38..db4395ea 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -48,6 +48,7 @@ MAX_PROC_PROCESS_SCAN_SIZE = 32_768 MAX_SEMANTIC_LOCATOR_CANDIDATES = 128 MAX_AGENT_TASK_STRUCTURED_VALUE_BYTES = 4_096 +MAX_AGENT_TASK_SEMANTIC_OBSERVATION_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 + "-_.") @@ -82,7 +83,7 @@ def _path_token(value: str, label: str) -> str: def _webdriver_path(session_id: str, suffix: str) -> str: - """Build one bounded ChromeDriver path from a validated session identifier.""" + """Build a bounded ChromeDriver path from a validated session identifier.""" safe_session = _path_token(session_id, "session identifier") if suffix and not suffix.startswith("/"): @@ -264,6 +265,24 @@ def _hash_agent_task_structured_value(value: str) -> str: return "sha256:" + hashlib.sha256(encoded).hexdigest() +def _measure_agent_task_semantic_observation_bytes(observation: dict[str, Any]) -> int: + """Measure one non-empty semantic observation under the canonical evidence bound.""" + + if not isinstance(observation, dict): + raise TypeError("Agent Task semantic observation must be an object") + if not observation: + raise ValueError("Agent Task semantic observation must not be empty") + encoded = json.dumps( + observation, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + if len(encoded) > MAX_AGENT_TASK_SEMANTIC_OBSERVATION_BYTES: + raise ValueError("Agent Task semantic observation exceeded the bounded evidence contract") + return len(encoded) + + def _parse_linux_proc_status_rss_bytes(status_text: str) -> int: """Parse exactly one positive Linux ``VmRSS`` kB field into bounded bytes.""" @@ -828,16 +847,9 @@ def _run_agent_task_browser_pass( "input": {"role": input_role, "name": input_name}, "submit": {"role": submit_role, "name": submit_name}, } - semantic_observation_bytes = len( - json.dumps( - semantic_observation, - ensure_ascii=False, - separators=(",", ":"), - sort_keys=True, - ).encode("utf-8") + semantic_observation_bytes = _measure_agent_task_semantic_observation_bytes( + semantic_observation ) - if semantic_observation_bytes <= 0: - raise RuntimeError("Agent Task semantic observation was empty") action_started = time.monotonic() _json_request( @@ -1136,7 +1148,9 @@ def main() -> int: and isinstance(trial.get("chromium_process_set_rss_bytes"), int) and trial["chromium_process_set_rss_bytes"] > 0 and isinstance(trial.get("semantic_observation_bytes"), int) - and trial["semantic_observation_bytes"] > 0 + and 0 + < trial["semantic_observation_bytes"] + <= MAX_AGENT_TASK_SEMANTIC_OBSERVATION_BYTES and isinstance(trial.get("action_latency_ms"), (int, float)) and trial["action_latency_ms"] > 0 and isinstance(trial.get("task_duration_ms"), (int, float)) diff --git a/tests/test_agent_task_observation_bound_contract.py b/tests/test_agent_task_observation_bound_contract.py new file mode 100644 index 00000000..fd46021e --- /dev/null +++ b/tests/test_agent_task_observation_bound_contract.py @@ -0,0 +1,60 @@ +"""Contract for bounded semantic-observation evidence in the controlled Agent Task.""" + +from __future__ import annotations + +import pathlib +import runpy +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class AgentTaskObservationBoundContractTests(unittest.TestCase): + """Require the pinned-browser Agent Task to fail closed on oversized observations.""" + + @classmethod + def setUpClass(cls) -> None: + cls.namespace = runpy.run_path( + str(RUNNER), run_name="agent_task_observation_bound_contract" + ) + + def test_semantic_observation_has_an_explicit_byte_limit(self) -> None: + """The runner must expose one finite semantic-observation byte ceiling.""" + + self.assertIn("MAX_AGENT_TASK_SEMANTIC_OBSERVATION_BYTES", self.namespace) + maximum = self.namespace["MAX_AGENT_TASK_SEMANTIC_OBSERVATION_BYTES"] + self.assertIsInstance(maximum, int) + self.assertGreater(maximum, 0) + self.assertLessEqual(maximum, 64 * 1024) + + def test_observation_measurement_accepts_exact_limit_and_rejects_overflow(self) -> None: + """Canonical UTF-8 evidence at the ceiling is valid; one byte over fails closed.""" + + self.assertIn("_measure_agent_task_semantic_observation_bytes", self.namespace) + helper = self.namespace["_measure_agent_task_semantic_observation_bytes"] + maximum = self.namespace["MAX_AGENT_TASK_SEMANTIC_OBSERVATION_BYTES"] + + # Canonical compact JSON for {"x":"..."} uses exactly eight structural bytes. + exact = {"x": "a" * (maximum - 8)} + oversized = {"x": "a" * (maximum - 7)} + self.assertEqual(helper(exact), maximum) + with self.assertRaises(ValueError): + helper(oversized) + with self.assertRaises(ValueError): + helper({}) + with self.assertRaises(TypeError): + helper("not-an-observation") + + def test_real_agent_task_path_uses_the_bounded_measurement_helper(self) -> None: + """The real controlled browser pass must not bypass the bounded helper.""" + + runner = RUNNER.read_text(encoding="utf-8") + self.assertIn( + "semantic_observation_bytes = _measure_agent_task_semantic_observation_bytes(", + runner, + ) + + +if __name__ == "__main__": + unittest.main()