test(browser): measure bounded Chromium process-tree RSS

CHANGELOG.md

@@ -23,6 +23,7 @@ All notable changes to OriginWeave are documented in this file. The format follo
 - Cumulative interactive-first RAM, VRAM, batch, local-model, admission, pause, and compositor-pressure mitigation plans, including active-consumer reduction at exact hard limits.
 - Universally value-redacted network evidence with explicit path, metadata, and provenance bounds; ambiguous path rejection; validated source URLs; lowercase SHA-256 identifiers; and verification state.
 - 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.
 - 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.

scripts/ci/run_mv3_compatibility.py

@@ -42,6 +42,8 @@
 FIXTURE_TIMEOUT_SECONDS = 20.0
 MAX_WEBDRIVER_RESPONSE_BYTES = 1_048_576
 MAX_PROC_STATUS_CHARACTERS = 65_536
+MAX_BROWSER_PROCESS_TREE_SIZE = 256
+MAX_PROC_PROCESS_SCAN_SIZE = 32_768
 MAX_U64 = (1 << 64) - 1
 W3C_ELEMENT_KEY = "element-6066-11e4-a52e-4f735466cecf"
 PATH_TOKEN_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_.")
@@ -76,7 +78,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("/"):
@@ -228,6 +230,58 @@ def _parse_linux_proc_status_rss_bytes(status_text: str) -> int:
     return rss_values[0]
 
 
+def _parse_linux_proc_status_optional_rss_bytes(status_text: str) -> int | None:
+    """Parse optional Linux ``VmRSS`` without normalizing malformed evidence."""
+
+    rss_lines = [line for line in status_text.splitlines() if line.startswith("VmRSS:")]
+    if not rss_lines:
+        return None
+    if len(rss_lines) != 1:
+        raise ValueError("Linux proc status must contain at most one VmRSS field")
+
+    fields = rss_lines[0].split()
+    if len(fields) != 3 or fields[0] != "VmRSS:" or fields[2] != "kB":
+        raise ValueError("malformed Linux VmRSS field")
+    raw_kibibytes = fields[1]
+    if not raw_kibibytes.isascii() or not raw_kibibytes.isdigit():
+        raise ValueError("malformed Linux VmRSS value")
+    kibibytes = int(raw_kibibytes, 10)
+    if kibibytes == 0:
+        return None
+    if kibibytes > MAX_U64 // 1024:
+        raise OverflowError("Linux VmRSS exceeds u64 byte range")
+    return kibibytes * 1024
+
+
+def _parse_linux_proc_status_process_identity(status_text: str) -> tuple[int, int]:
+    """Parse exactly one positive ``Pid`` and one non-negative ``PPid`` from status."""
+
+    parsed: dict[str, int] = {}
+    for line in status_text.splitlines():
+        if not (line.startswith("Pid:") or line.startswith("PPid:")):
+            continue
+        fields = line.split()
+        if len(fields) != 2 or fields[0] not in {"Pid:", "PPid:"}:
+            raise ValueError("malformed Linux process identity field")
+        label = fields[0]
+        if label in parsed:
+            raise ValueError("duplicate Linux process identity field")
+        raw_process_id = fields[1]
+        if not raw_process_id.isascii() or not raw_process_id.isdigit():
+            raise ValueError("malformed Linux process identity value")
+        parsed[label] = int(raw_process_id, 10)
+
+    if set(parsed) != {"Pid:", "PPid:"}:
+        raise ValueError("Linux proc status must contain exactly one Pid and PPid")
+    process_id = parsed["Pid:"]
+    parent_process_id = parsed["PPid:"]
+    if process_id <= 0:
+        raise ValueError("Linux process identifier must be positive")
+    if parent_process_id < 0:
+        raise ValueError("Linux parent process identifier must be non-negative")
+    return process_id, parent_process_id
+
+
 def _sample_linux_process_rss_bytes(process_id: int) -> int:
     """Read one attributed Linux process RSS through a bounded ``/proc`` status file."""
 
@@ -241,6 +295,105 @@ def _sample_linux_process_rss_bytes(process_id: int) -> int:
     return _parse_linux_proc_status_rss_bytes(status_text)
 
 
+def _snapshot_linux_process_evidence() -> dict[int, tuple[int, int | None]]:
+    """Capture one bounded best-effort PID/PPID/RSS sweep from Linux proc status."""
+
+    proc_root = pathlib.Path("/proc")
+    process_entries: list[tuple[int, pathlib.Path]] = []
+    for entry in proc_root.iterdir():
+        raw_process_id = entry.name
+        if not raw_process_id.isascii() or not raw_process_id.isdigit():
+            continue
+        process_id = int(raw_process_id, 10)
+        if process_id <= 0:
+            continue
+        process_entries.append((process_id, entry))
+        if len(process_entries) > MAX_PROC_PROCESS_SCAN_SIZE:
+            raise RuntimeError("Linux proc process scan exceeded the bounded entry limit")
+
+    process_evidence: dict[int, tuple[int, int | None]] = {}
+    for expected_process_id, entry in sorted(process_entries):
+        status_path = entry / "status"
+        try:
+            with status_path.open("r", encoding="utf-8", errors="strict") as status_file:
+                status_text = status_file.read(MAX_PROC_STATUS_CHARACTERS + 1)
+        except FileNotFoundError:
+            continue
+        if len(status_text) > MAX_PROC_STATUS_CHARACTERS:
+            raise RuntimeError("Linux proc status exceeded the bounded text limit")
+        process_id, parent_process_id = _parse_linux_proc_status_process_identity(
+            status_text
+        )
+        if process_id != expected_process_id:
+            raise RuntimeError("Linux proc status identity did not match its directory")
+        if process_id in process_evidence:
+            raise RuntimeError("Linux proc process snapshot contained a duplicate PID")
+        rss_bytes = _parse_linux_proc_status_optional_rss_bytes(status_text)
+        process_evidence[process_id] = (parent_process_id, rss_bytes)
+    return process_evidence
+
+
+def _discover_linux_process_tree_ids(
+    root_process_id: int,
+    process_evidence: dict[int, tuple[int, int | None]],
+) -> tuple[int, ...]:
+    """Discover one bounded root-plus-descendant set from sampled process evidence."""
+
+    if (
+        isinstance(root_process_id, bool)
+        or not isinstance(root_process_id, int)
+        or root_process_id <= 0
+    ):
+        raise ValueError("invalid Linux root process identifier")
+    if root_process_id not in process_evidence:
+        raise RuntimeError("Linux process snapshot did not contain the browser root PID")
+
+    discovered = [root_process_id]
+    known = {root_process_id}
+    while True:
+        children = sorted(
+            process_id
+            for process_id, (parent_process_id, _rss_bytes) in process_evidence.items()
+            if parent_process_id in known and process_id not in known
+        )
+        if not children:
+            break
+        for process_id in children:
+            if len(known) >= MAX_BROWSER_PROCESS_TREE_SIZE:
+                raise ValueError("Linux process tree exceeded the bounded process-tree size")
+            known.add(process_id)
+            discovered.append(process_id)
+    return tuple(discovered)
+
+
+def _sample_linux_process_set_rss_bytes(
+    process_ids: tuple[int, ...],
+    process_evidence: dict[int, tuple[int, int | None]],
+) -> int:
+    """Sum resident RSS for one exact bounded process set without overflow."""
+
+    if not process_ids or len(process_ids) > MAX_BROWSER_PROCESS_TREE_SIZE:
+        raise ValueError("invalid Linux process set size")
+    if len(set(process_ids)) != len(process_ids):
+        raise ValueError("Linux process set identifiers must be unique")
+
+    total_rss_bytes = 0
+    for process_id in process_ids:
+        if isinstance(process_id, bool) or not isinstance(process_id, int) or process_id <= 0:
+            raise ValueError("invalid Linux process identifier")
+        if process_id not in process_evidence:
+            raise ValueError("Linux process set was not present in the sampled evidence")
+        rss_bytes = process_evidence[process_id][1]
+        if rss_bytes is None:
+            continue
+        if isinstance(rss_bytes, bool) or not isinstance(rss_bytes, int) or rss_bytes <= 0:
+            raise ValueError("Linux process set contained invalid sampled RSS")
+        if rss_bytes > MAX_U64 - total_rss_bytes:
+            raise OverflowError("Linux process-set RSS exceeds u64 byte range")
+        total_rss_bytes += rss_bytes
+    return total_rss_bytes
+
+
 def _wait_for_extension_evidence(
     driver_port: int,
     session_id: str,
@@ -652,7 +805,18 @@ def _run_agent_task_browser_pass(
             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")
+
+        process_evidence = _snapshot_linux_process_evidence()
+        chromium_process_ids = _discover_linux_process_tree_ids(
+            browser_process_id,
+            process_evidence,
+        )
         browser_process_rss_bytes = _sample_linux_process_rss_bytes(browser_process_id)
+        chromium_process_set_rss_bytes = _sample_linux_process_set_rss_bytes(
+            chromium_process_ids,
+            process_evidence,
+        )
+        chromium_process_count = len(chromium_process_ids)
         task_duration_ms = round((time.monotonic() - started) * 1000, 3)
         if task_duration_ms <= 0:
             raise RuntimeError("Agent Task measured a non-positive task duration")
@@ -664,6 +828,8 @@ def _run_agent_task_browser_pass(
             "submit_semantics_verified": True,
             "extensions_disabled": True,
             "browser_process_rss_bytes": browser_process_rss_bytes,
+            "chromium_process_count": chromium_process_count,
+            "chromium_process_set_rss_bytes": chromium_process_set_rss_bytes,
             "semantic_observation_bytes": semantic_observation_bytes,
             "action_latency_ms": action_latency_ms,
             "task_duration_ms": task_duration_ms,
@@ -720,6 +886,8 @@ def _run_agent_task_trial(
         "submit_semantics_verified": result["submit_semantics_verified"],
         "extensions_disabled": result["extensions_disabled"],
         "browser_process_rss_bytes": result["browser_process_rss_bytes"],
+        "chromium_process_count": result["chromium_process_count"],
+        "chromium_process_set_rss_bytes": result["chromium_process_set_rss_bytes"],
         "semantic_observation_bytes": result["semantic_observation_bytes"],
         "action_latency_ms": result["action_latency_ms"],
         "task_duration_ms": result["task_duration_ms"],
@@ -788,11 +956,12 @@ def main() -> int:
                         trial_number,
                     )
                 )
-            except (OSError, ValueError, RuntimeError, json.JSONDecodeError):
+            except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
                 trial_results.append(
                     {
                         "trial_number": trial_number,
                         "passed": False,
+                        "failure_type": type(exc).__name__,
                     }
                 )
 
@@ -830,11 +999,12 @@ def main() -> int:
                         trial_number,
                     )
                 )
-            except (OSError, ValueError, RuntimeError, json.JSONDecodeError):
+            except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
                 agent_task_trials.append(
                     {
                         "trial_number": trial_number,
                         "passed": False,
+                        "failure_type": type(exc).__name__,
                     }
                 )
 
@@ -853,6 +1023,10 @@ def main() -> int:
             and trial.get("profile_cleaned") is True
             and isinstance(trial.get("browser_process_rss_bytes"), int)
             and trial["browser_process_rss_bytes"] > 0
+            and isinstance(trial.get("chromium_process_count"), int)
+            and 0 < trial["chromium_process_count"] <= MAX_BROWSER_PROCESS_TREE_SIZE
+            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 isinstance(trial.get("action_latency_ms"), (int, float))

tests/test_agent_task_fixture_contract.py

@@ -10,6 +10,21 @@
 FIXTURE = ROOT / "tests" / "fixtures" / "agent_task_basic" / "index.html"
 
 
+def _is_credential_input(attributes: dict[str, str | None]) -> bool:
+    """Return whether parsed input attributes describe a credential surface."""
+
+    input_type = (attributes.get("type") or "").strip().lower()
+    if input_type == "password":
+        return True
+
+    autocomplete = (attributes.get("autocomplete") or "").strip().lower()
+    autocomplete_tokens = autocomplete.split()
+    return any(
+        token == "one-time-code" or "password" in token
+        for token in autocomplete_tokens
+    )
+
+
 class _FixtureParser(HTMLParser):
     """Collect the small semantic surface required by the deterministic fixture."""
 
@@ -18,6 +33,7 @@ def __init__(self) -> None:
         self.ids: set[str] = set()
         self.labels_for: set[str] = set()
         self.input_names: set[str] = set()
+        self.input_attributes: list[dict[str, str | None]] = []
         self.button_types: set[str] = set()
         self.hidden_injection_markers = 0
 
@@ -30,12 +46,15 @@ def handle_starttag(
             self.ids.add(element_id)
         if tag == "label" and attributes.get("for"):
             self.labels_for.add(attributes["for"])
-        if tag == "input" and attributes.get("name"):
-            self.input_names.add(attributes["name"])
+        if tag == "input":
+            self.input_attributes.append(attributes)
+            if attributes.get("name"):
+                self.input_names.add(attributes["name"])
         if tag == "button" and attributes.get("type"):
             self.button_types.add(attributes["type"])
         if (
             attributes.get("data-originweave-untrusted") == "prompt-injection"
+            and "hidden" in attributes
             and attributes.get("aria-hidden") == "true"
         ):
             self.hidden_injection_markers += 1
@@ -70,20 +89,49 @@ def test_fixture_contains_explicit_untrusted_hidden_prompt_injection(self) -> No
         self.assertIn("UNTRUSTED_PAGE_INSTRUCTION", self.html)
         self.assertIn("request new browser capabilities", self.html)
 
+    def test_hidden_injection_requires_the_actual_hidden_attribute(self) -> None:
+        """ARIA metadata alone must not satisfy the hidden-injection fixture contract."""
+
+        parser = _FixtureParser()
+        parser.feed(
+            "<p data-originweave-untrusted='prompt-injection' aria-hidden='true'>visible</p>"
+            "<p data-originweave-untrusted='prompt-injection' aria-hidden='true' hidden>hidden</p>"
+        )
+        self.assertEqual(parser.hidden_injection_markers, 1)
+
     def test_fixture_is_synthetic_and_has_no_credential_fields(self) -> None:
         """The controlled workflow must not require or imitate real secret collection."""
 
+        for attributes in self.parser.input_attributes:
+            with self.subTest(attributes=attributes):
+                self.assertFalse(_is_credential_input(attributes))
+
         lowered = self.html.lower()
-        for forbidden in (
-            'type="password"',
-            'autocomplete="current-password"',
-            'autocomplete="one-time-code"',
-            "api_key",
-            "secret_key",
-        ):
+        for forbidden in ("api_key", "secret_key"):
             with self.subTest(forbidden=forbidden):
                 self.assertNotIn(forbidden, lowered)
 
+    def test_credential_detection_is_quote_independent(self) -> None:
+        """Parsed credential semantics must reject single-quoted and tokenized forms."""
+
+        for html in (
+            "<input type='password'>",
+            "<input autocomplete='current-password'>",
+            "<input autocomplete='new-password'>",
+            "<input autocomplete='section-login username current-password'>",
+            "<input autocomplete='one-time-code'>",
+        ):
+            with self.subTest(html=html):
+                parser = _FixtureParser()
+                parser.feed(html)
+                self.assertEqual(len(parser.input_attributes), 1)
+                self.assertTrue(_is_credential_input(parser.input_attributes[0]))
+
+        parser = _FixtureParser()
+        parser.feed("<input type='text' autocomplete='username'>")
+        self.assertEqual(len(parser.input_attributes), 1)
+        self.assertFalse(_is_credential_input(parser.input_attributes[0]))
+
 
 if __name__ == "__main__":
     unittest.main()