feat(browser): fail closed on oversized Agent Task observation

CHANGELOG.md

@@ -25,6 +25,7 @@ All notable changes to OriginWeave are documented in this file. The format follo
 - Rust 1.97.1 build contract, strict Clippy and rustdoc gates, and exact production function, line, region, and branch coverage enforcement.
 - Pinned Chrome-for-Testing Agent Task evidence now captures a bounded sampled Chromium root-plus-descendant process count and RSS total from one `/proc` status sweep, with bounded failure-type diagnostics while preserving the root-only metric and making no trusted per-task attribution claim.
 - Pinned Chrome-for-Testing Agent Task evidence now locates the controlled result by exact browser-computed `status`/`Task result` semantics and records only a bounded canonical SHA-256 digest plus stable field identity for the extracted synthetic value, without emitting the raw value.
+- Pinned Chrome-for-Testing Agent Task semantic-observation evidence is now canonicalized as compact sorted-key UTF-8 JSON and capped at 4,096 bytes, accepting the exact limit while failing closed on empty, non-object, or oversized observations before they enter successful trial evidence.
 - Hourly bounded OpenCode product-development workflow using `NVIDIA_NIM_API_KEY`, an unprivileged disposable workspace, loopback-only model broker, independently verified patches, and publication through a dedicated `OPENCODE_PR_TOKEN` that cannot review or merge.
 - Architecture, agent, security, contribution, research, database naming, roadmap, quality-gate, and TLS service-identity ADR documentation.
 - Authoritative product documentation graph spanning PRD, TRD, ADR lifecycle/index, product-wide UML, conceptual ERD, requirement/decision traceability, threat modeling, product-wide test strategy, operability, API/protocol, release/rollback, and current primary-source standards doctoring, with machine-checkable repository contracts that keep conversation-derived future work distinct from protected-main implementation claims.
@@ -47,6 +48,7 @@ All notable changes to OriginWeave are documented in this file. The format follo
 ### Security
 
 - Raw page content cannot become a trusted instruction.
+- Controlled Agent Task compatibility evidence is machine-checked to contain only the reviewed `input` and `submit` browser-computed role/name fields before bounded measurement, preventing unreviewed page text or instruction-like fields from silently entering that evidence object.
 - Raw secrets are rejected and secret-capable actions require an opaque broker handle.
 - Crawler mode is read-only, must pair with the public-crawl purpose, and fails closed without an applicable robots-policy decision.
 - State-changing actions are same-origin by default.
@@ -75,4 +77,4 @@ All notable changes to OriginWeave are documented in this file. The format follo
 - The hourly product agent has no Git metadata or repository authority. A separate post-verification publisher opens one PR and cannot approve or merge it.
 - The unprivileged OpenCode user is restricted to loopback egress during model execution, preventing runner-wide allow-listed endpoints from becoming direct source-exfiltration channels.
 
-[Unreleased]: https://github.com/ContextualWisdomLab/OriginWeave/compare/main...HEAD
+[Unreleased]: https://github.com/ContextualWisdomLab/OriginWeave/compare/main...HEAD

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))

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()

tests/test_agent_task_semantic_observation_schema_contract.py

@@ -0,0 +1,90 @@
+"""Contract for the controlled Agent Task semantic-observation evidence schema."""
+
+from __future__ import annotations
+
+import ast
+import pathlib
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py"
+
+
+class AgentTaskSemanticObservationSchemaContractTests(unittest.TestCase):
+    """Keep unreviewed page content outside the controlled semantic evidence object."""
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.tree = ast.parse(RUNNER.read_text(encoding="utf-8"), filename=str(RUNNER))
+
+    @staticmethod
+    def _literal_dict_keys(node: ast.Dict) -> tuple[str, ...]:
+        """Return exact string-literal dictionary keys or fail the contract."""
+
+        keys: list[str] = []
+        for key in node.keys:
+            if not isinstance(key, ast.Constant) or not isinstance(key.value, str):
+                raise AssertionError("semantic observation keys must be string literals")
+            keys.append(key.value)
+        return tuple(keys)
+
+    def _semantic_observation_assignment(self) -> ast.Dict:
+        """Find the one executable controlled-observation construction."""
+
+        assignments = [
+            node
+            for node in ast.walk(self.tree)
+            if isinstance(node, ast.Assign)
+            and any(
+                isinstance(target, ast.Name) and target.id == "semantic_observation"
+                for target in node.targets
+            )
+        ]
+        self.assertEqual(len(assignments), 1)
+        value = assignments[0].value
+        self.assertIsInstance(value, ast.Dict)
+        return value
+
+    def test_controlled_observation_has_only_reviewed_role_name_fields(self) -> None:
+        """Raw page text or instruction-like fields cannot drift into emitted evidence."""
+
+        observation = self._semantic_observation_assignment()
+        self.assertEqual(self._literal_dict_keys(observation), ("input", "submit"))
+        semantic_nodes = dict(zip(self._literal_dict_keys(observation), observation.values))
+
+        expected_values = {
+            "input": {"role": "input_role", "name": "input_name"},
+            "submit": {"role": "submit_role", "name": "submit_name"},
+        }
+        for semantic_key, expected_fields in expected_values.items():
+            semantic_node = semantic_nodes[semantic_key]
+            self.assertIsInstance(semantic_node, ast.Dict)
+            self.assertEqual(self._literal_dict_keys(semantic_node), ("role", "name"))
+            actual_fields = dict(
+                zip(self._literal_dict_keys(semantic_node), semantic_node.values)
+            )
+            for field_name, expected_variable in expected_fields.items():
+                value = actual_fields[field_name]
+                self.assertIsInstance(value, ast.Name)
+                self.assertEqual(value.id, expected_variable)
+
+    def test_exact_observation_flows_through_bounded_measurement(self) -> None:
+        """The reviewed object must be the exact object sent to the byte-bound helper."""
+
+        calls = [
+            node
+            for node in ast.walk(self.tree)
+            if isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Name)
+            and node.func.id == "_measure_agent_task_semantic_observation_bytes"
+        ]
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(len(calls[0].args), 1)
+        argument = calls[0].args[0]
+        self.assertIsInstance(argument, ast.Name)
+        self.assertEqual(argument.id, "semantic_observation")
+        self.assertFalse(calls[0].keywords)
+
+
+if __name__ == "__main__":
+    unittest.main()