test(performance): add deterministic scale baselines

scripts/verify_static_analysis_policy.py

@@ -66,7 +66,9 @@ def _repo_relative_path(path_value: Any, repo_root: Path) -> str:
     try:
         relative = path.resolve().relative_to(repo_root.resolve())
     except ValueError as exc:
-        raise ValueError(f"diagnostic path is outside repository: {path_value}") from exc
+        raise ValueError(
+            f"diagnostic path is outside repository: {path_value}"
+        ) from exc
     return relative.as_posix()
 
 
@@ -323,9 +325,7 @@ def _run_command(
 
 
 def _parse_tool_version(tool_name: str, output: str) -> str:
-    match = re.search(
-        rf"\b{re.escape(tool_name)}\s+([0-9]+(?:\.[0-9]+)+)", output
-    )
+    match = re.search(rf"\b{re.escape(tool_name)}\s+([0-9]+(?:\.[0-9]+)+)", output)
     if not match:
         raise ToolExecutionError(f"{tool_name} version output was not recognized")
     return match.group(1)
@@ -546,9 +546,7 @@ def main(argv: list[str] | None = None) -> int:
         policy = json.loads(policy_path.read_text(encoding="utf-8"))
         requirement_lines = requirements_path.read_text(encoding="utf-8").splitlines()
         structural_failures = validate_policy(repo_root, policy)
-        structural_failures.extend(
-            validate_requirement_pins(policy, requirement_lines)
-        )
+        structural_failures.extend(validate_requirement_pins(policy, requirement_lines))
         if structural_failures:
             for failure in structural_failures:
                 print(f"STATIC-ANALYSIS-FAIL: {failure}")
@@ -589,7 +587,9 @@ def main(argv: list[str] | None = None) -> int:
         )
         return 0
     except (OSError, ValueError, json.JSONDecodeError) as exc:
-        print(f"STATIC-ANALYSIS-FAIL: invalid policy or tool output ({type(exc).__name__})")
+        print(
+            f"STATIC-ANALYSIS-FAIL: invalid policy or tool output ({type(exc).__name__})"
+        )
         return 1
     except ToolExecutionError as exc:
         print(f"STATIC-ANALYSIS-FAIL: {exc}")

tests/performance_baseline_policy.json

@@ -0,0 +1,74 @@
+{
+  "schema_version": 1,
+  "policy_id": "openclaw-deterministic-scale-v1",
+  "reviewed_on": "2026-07-11",
+  "timing": {
+    "clock": "monotonic_high_resolution",
+    "enforcement": "advisory_only",
+    "samples": 2
+  },
+  "workloads": [
+    {
+      "id": "backend_jobs_history",
+      "seed": 21801,
+      "owner": "services.jobs_read_model",
+      "review_after": "2027-01-11",
+      "input": {
+        "history_records": 10001
+      },
+      "expected": {
+        "exact": {
+          "source_records": 10001,
+          "examined": 10000,
+          "total": 10000,
+          "returned": 50,
+          "truncated": true,
+          "queue_snapshot_calls": 1,
+          "history_snapshot_calls": 1,
+          "upstream_calls": 1,
+          "upstream_records": 10000
+        },
+        "max_payload_bytes": 100000,
+        "digest_sha256": "6fdc64ed424a242c6d9a0cba5761440a1789393e2d1036001fbe3fd60f98bde5"
+      }
+    },
+    {
+      "id": "connector_jobs_dispatch",
+      "seed": 21802,
+      "owner": "connector.jobs_summary",
+      "review_after": "2027-01-11",
+      "input": {
+        "returned_jobs": 200
+      },
+      "expected": {
+        "exact": {
+          "returned_jobs": 200,
+          "client_calls": 1,
+          "fallback_calls": 0,
+          "visible_job_lines": 5
+        },
+        "max_summary_chars": 1000,
+        "digest_sha256": "275086b759bff79c7b0ce54505bf10a60e82ae8966016ef4bafb3727e9684863"
+      }
+    },
+    {
+      "id": "frontend_history_outputs",
+      "seed": 21803,
+      "owner": "web.openclaw_asset_refs",
+      "review_after": "2027-01-11",
+      "input": {
+        "nodes": 256,
+        "refs_per_node": 4
+      },
+      "expected": {
+        "exact": {
+          "input_refs": 1024,
+          "normalized_outputs": 1024,
+          "image_outputs": 1024
+        },
+        "max_serialized_bytes": 500000,
+        "digest_sha256": "cc6ce5107a595bf1a6d509918a339c5e2bc228b014e7281816b2c888037a08ea"
+      }
+    }
+  ]
+}

tests/test_r217_static_analysis_policy.py

@@ -46,7 +46,9 @@ class TestStaticAnalysisPolicy(unittest.TestCase):
         (root / "pkg" / "nested").mkdir(parents=True)
         (root / "pkg" / "owned.py").write_text("import os\n", encoding="utf-8")
         (root / "pkg" / "clean.py").write_text("VALUE: int = 1\n", encoding="utf-8")
-        (root / "pkg" / "generated.py").write_text("generated = True\n", encoding="utf-8")
+        (root / "pkg" / "generated.py").write_text(
+            "generated = True\n", encoding="utf-8"
+        )
         (root / "pkg" / "nested" / "child.py").write_text(
             "CHILD = True\n", encoding="utf-8"
         )
@@ -373,9 +375,9 @@ class TestStaticAnalysisPolicy(unittest.TestCase):
                     name = command[2]
                     return SimpleNamespace(
                         returncode=0,
-                        stdout=f"{name} 0.15.20\n"
-                        if name == "ruff"
-                        else "mypy 2.2.0\n",
+                        stdout=(
+                            f"{name} 0.15.20\n" if name == "ruff" else "mypy 2.2.0\n"
+                        ),
                         stderr="",
                     )
                 return SimpleNamespace(
@@ -414,9 +416,7 @@ class TestStaticAnalysisPolicy(unittest.TestCase):
                     return SimpleNamespace(
                         returncode=0,
                         stdout=(
-                            "ruff 0.15.19\n"
-                            if command[2] == "ruff"
-                            else "mypy 2.2.0\n"
+                            "ruff 0.15.19\n" if command[2] == "ruff" else "mypy 2.2.0\n"
                         ),
                         stderr="",
                     )
@@ -438,9 +438,7 @@ class TestStaticAnalysisPolicy(unittest.TestCase):
             "ruff requirement drift: expected ruff==0.15.20, found ruff>=0.15.20",
             failures,
         )
-        self.assertIn(
-            "ruff version drift: expected 0.15.20, found 0.15.19", failures
-        )
+        self.assertIn("ruff version drift: expected 0.15.20, found 0.15.19", failures)
 
 
 class TestRepositoryStaticAnalysisPolicy(unittest.TestCase):
@@ -452,7 +450,9 @@ class TestRepositoryStaticAnalysisPolicy(unittest.TestCase):
 
     def test_repository_policy_and_quality_pins_are_valid(self):
         policy = json.loads(self.policy_path.read_text(encoding="utf-8"))
-        requirement_lines = self.requirements_path.read_text(encoding="utf-8").splitlines()
+        requirement_lines = self.requirements_path.read_text(
+            encoding="utf-8"
+        ).splitlines()
 
         self.assertEqual(policy_module.validate_policy(self.repo_root, policy), [])
         self.assertEqual(
@@ -471,16 +471,23 @@ class TestRepositoryStaticAnalysisPolicy(unittest.TestCase):
                 "scripts",
             ],
         )
-        self.assertGreater(len(policy_module.discover_owned_python_files(self.repo_root, policy)), 200)
+        self.assertGreater(
+            len(policy_module.discover_owned_python_files(self.repo_root, policy)), 200
+        )
         self.assertGreaterEqual(len(policy["strict_paths"]), 1)
 
     def test_repository_execution_surfaces_use_shared_verifier(self):
         expected = "scripts/verify_static_analysis_policy.py"
         surfaces = {
             ".pre-commit-config.yaml": self.repo_root / ".pre-commit-config.yaml",
-            "windows full gate": self.repo_root / "scripts" / "run_full_tests_windows.ps1",
+            "windows full gate": self.repo_root
+            / "scripts"
+            / "run_full_tests_windows.ps1",
             "linux full gate": self.repo_root / "scripts" / "run_full_tests_linux.sh",
-            "pre-commit CI": self.repo_root / ".github" / "workflows" / "pre-commit.yml",
+            "pre-commit CI": self.repo_root
+            / ".github"
+            / "workflows"
+            / "pre-commit.yml",
             "unit CI": self.repo_root / ".github" / "workflows" / "ci.yml",
         }
         for label, path in surfaces.items():

tests/test_r218_performance_baseline.py

@@ -0,0 +1,429 @@
+"""Deterministic scale baselines for jobs and connector hot paths."""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import hashlib
+import json
+import random
+import re
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from connector.config import ConnectorConfig
+from connector.contract import CommandRequest
+from connector.router import CommandRouter
+from services import jobs_read_model
+from services.jobs_security import normalize_jobs_query
+
+ROOT = Path(__file__).resolve().parents[1]
+POLICY_PATH = ROOT / "tests" / "performance_baseline_policy.json"
+EXPECTED_WORKLOAD_IDS = {
+    "backend_jobs_history",
+    "connector_jobs_dispatch",
+    "frontend_history_outputs",
+}
+_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+
+
+def validate_performance_policy(policy: object) -> list[str]:
+    """Return deterministic schema, budget, and privacy violations."""
+
+    errors: list[str] = []
+    if not isinstance(policy, dict):
+        return ["policy must be an object"]
+    if set(policy) != {
+        "schema_version",
+        "policy_id",
+        "reviewed_on",
+        "timing",
+        "workloads",
+    }:
+        errors.append("root keys must match the versioned schema")
+    if policy.get("schema_version") != 1:
+        errors.append("schema_version must equal 1")
+
+    timing = policy.get("timing")
+    if not isinstance(timing, dict) or set(timing) != {
+        "clock",
+        "enforcement",
+        "samples",
+    }:
+        errors.append("timing keys must match the advisory schema")
+    elif (
+        timing.get("clock") != "monotonic_high_resolution"
+        or timing.get("enforcement") != "advisory_only"
+        or timing.get("samples") != 2
+    ):
+        errors.append("timing must be two advisory monotonic samples")
+
+    workloads = policy.get("workloads")
+    if not isinstance(workloads, list):
+        errors.append("workloads must be a list")
+        workloads = []
+    ids = [item.get("id") for item in workloads if isinstance(item, dict)]
+    if set(ids) != EXPECTED_WORKLOAD_IDS or len(ids) != len(set(ids)):
+        errors.append("workload ids must be unique and complete")
+
+    expected_inputs = {
+        "backend_jobs_history": ({"history_records"}, 10_001),
+        "connector_jobs_dispatch": ({"returned_jobs"}, 200),
+        "frontend_history_outputs": ({"nodes", "refs_per_node"}, 512),
+    }
+    expected_outputs = {
+        "backend_jobs_history": (
+            {
+                "source_records",
+                "examined",
+                "total",
+                "returned",
+                "truncated",
+                "queue_snapshot_calls",
+                "history_snapshot_calls",
+                "upstream_calls",
+                "upstream_records",
+            },
+            "max_payload_bytes",
+        ),
+        "connector_jobs_dispatch": (
+            {"returned_jobs", "client_calls", "fallback_calls", "visible_job_lines"},
+            "max_summary_chars",
+        ),
+        "frontend_history_outputs": (
+            {"input_refs", "normalized_outputs", "image_outputs"},
+            "max_serialized_bytes",
+        ),
+    }
+    for item in workloads:
+        if not isinstance(item, dict) or set(item) != {
+            "id",
+            "seed",
+            "owner",
+            "review_after",
+            "input",
+            "expected",
+        }:
+            errors.append("workload keys must match the versioned schema")
+            continue
+        workload_id = item["id"]
+        seed = item["seed"]
+        if isinstance(seed, bool) or not isinstance(seed, int) or seed <= 0:
+            errors.append(f"{workload_id}: seed must be a positive integer")
+        if not isinstance(item["owner"], str) or not item["owner"]:
+            errors.append(f"{workload_id}: owner is required")
+        if not isinstance(item["review_after"], str) or not item["review_after"]:
+            errors.append(f"{workload_id}: review date is required")
+
+        inputs = item["input"]
+        input_schema = expected_inputs.get(workload_id)
+        if (
+            not isinstance(inputs, dict)
+            or input_schema is None
+            or set(inputs) != input_schema[0]
+        ):
+            errors.append(f"{workload_id}: input keys are invalid")
+        else:
+            for name, value in inputs.items():
+                if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+                    errors.append(f"{workload_id}: {name} must be positive")
+                elif value > input_schema[1]:
+                    errors.append(f"{workload_id}: {name} exceeds its safe test bound")
+            if workload_id == "frontend_history_outputs" and (
+                inputs["nodes"] * inputs["refs_per_node"] > 4096
+            ):
+                errors.append(
+                    "frontend_history_outputs: total refs exceed the safe test bound"
+                )
+
+        expected = item["expected"]
+        output_schema = expected_outputs.get(workload_id)
+        if (
+            not isinstance(expected, dict)
+            or output_schema is None
+            or set(expected) != {"exact", output_schema[1], "digest_sha256"}
+            or not isinstance(expected.get("exact"), dict)
+            or set(expected["exact"]) != output_schema[0]
+        ):
+            errors.append(f"{workload_id}: expected budgets are missing")
+        else:
+            digest = expected.get("digest_sha256")
+            if not isinstance(digest, str) or _SHA256.fullmatch(digest) is None:
+                errors.append(f"{workload_id}: canonical digest is invalid")
+            maxima = [
+                value for key, value in expected.items() if key.startswith("max_")
+            ]
+            if len(maxima) != 1 or any(
+                isinstance(value, bool) or not isinstance(value, int) or value <= 0
+                for value in maxima
+            ):
+                errors.append(f"{workload_id}: one positive maximum budget is required")
+
+    serialized = json.dumps(policy, sort_keys=True).lower()
+    for forbidden in (
+        "latency_threshold",
+        "max_seconds",
+        "b:\\",
+        "/home/",
+        "prompt",
+        "token",
+        "secret",
+    ):
+        if forbidden in serialized:
+            errors.append(f"policy contains forbidden content marker: {forbidden}")
+    return errors
+
+
+def _canonical_digest(value: object) -> str:
+    encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _load_policy() -> dict:
+    return json.loads(POLICY_PATH.read_text(encoding="utf-8"))
+
+
+def _workload(policy: dict, workload_id: str) -> dict:
+    return next(item for item in policy["workloads"] if item["id"] == workload_id)
+
+
+def _history_record(prompt_id: str, status: str) -> dict:
+    return {
+        "prompt": (0, prompt_id, {}, {"openclaw": {"tenant_id": "default"}}, []),
+        "synthetic_status": status,
+    }
+
+
+class _CountingPromptQueue:
+    def __init__(self, history: dict[str, dict]) -> None:
+        self.history = history
+        self.queue_calls = 0
+        self.history_calls = 0
+
+    def get_current_queue_volatile(self):
+        self.queue_calls += 1
+        return ([], [])
+
+    def get_history(self):
+        self.history_calls += 1
+        return self.history
+
+
+def _run_backend_probe(workload: dict) -> tuple[dict, float]:
+    size = workload["input"]["history_records"]
+    rng = random.Random(workload["seed"])
+    statuses = ("completed", "failed", "cancelled")
+    history = {
+        f"job-{index:05d}": _history_record(
+            f"job-{index:05d}", statuses[rng.randrange(len(statuses))]
+        )
+        for index in range(size)
+    }
+    queue = _CountingPromptQueue(history)
+    upstream_calls = 0
+    upstream_records = 0
+
+    def get_all_jobs(running, queued, bounded_history, **_kwargs):
+        nonlocal upstream_calls, upstream_records
+        upstream_calls += 1
+        upstream_records = len(running) + len(queued) + len(bounded_history)
+        jobs = [
+            {"id": prompt_id, "status": record["synthetic_status"]}
+            for prompt_id, record in bounded_history.items()
+        ]
+        return jobs, len(jobs)
+
+    started = time.perf_counter()
+    with (
+        patch.object(
+            jobs_read_model, "_resolve_get_all_jobs", return_value=get_all_jobs
+        ),
+        patch.object(jobs_read_model, "_resolve_prompt_queue", return_value=queue),
+        patch.object(jobs_read_model, "is_multi_tenant_enabled", return_value=False),
+    ):
+        body = jobs_read_model.read_jobs(normalize_jobs_query({}), tenant_id="default")
+    elapsed = time.perf_counter() - started
+    deterministic = {
+        "source_records": size,
+        "examined": body["scan"]["examined"],
+        "total": body["pagination"]["total"],
+        "returned": len(body["jobs"]),
+        "truncated": body["scan"]["truncated"],
+        "queue_snapshot_calls": queue.queue_calls,
+        "history_snapshot_calls": queue.history_calls,
+        "upstream_calls": upstream_calls,
+        "upstream_records": upstream_records,
+        "payload_bytes": len(
+            json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        ),
+        "digest": _canonical_digest(body),
+    }
+    return deterministic, elapsed
+
+
+def _connector_request() -> CommandRequest:
+    return CommandRequest(
+        platform="test",
+        sender_id="admin-user",
+        channel_id="scale-channel",
+        username="operator",
+        message_id="scale-message",
+        text="/jobs",
+        timestamp=0,
+    )
+
+
+def _run_connector_probe(workload: dict) -> tuple[dict, float]:
+    count = workload["input"]["returned_jobs"]
+    rng = random.Random(workload["seed"])
+    statuses = ("pending", "in_progress", "completed", "failed", "cancelled")
+    jobs = [
+        {"id": f"job-{index:03d}", "status": statuses[rng.randrange(len(statuses))]}
+        for index in range(count)
+    ]
+    response = {
+        "ok": True,
+        "status": 200,
+        "data": {
+            "ok": True,
+            "contract_version": 1,
+            "jobs": jobs,
+            "pagination": {
+                "offset": 0,
+                "limit": count,
+                "total": count,
+                "has_more": False,
+                "warnings": [],
+            },
+            "source": {"adapter": "comfy_execution.jobs", "authority": "in_process"},
+            "scan": {
+                "window": 10_000,
+                "examined": count,
+                "excluded": 0,
+                "malformed": 0,
+                "truncated": False,
+            },
+        },
+    }
+    config = ConnectorConfig()
+    config.admin_users = ["admin-user"]
+    config.admin_token = "configured-test-value"
+    client = MagicMock()
+    client.get_jobs = AsyncMock(return_value=response)
+    client.get_prompt_queue = AsyncMock()
+    router = CommandRouter(config, client)
+
+    started = time.perf_counter()
+    rendered = asyncio.run(router.handle(_connector_request())).text
+    elapsed = time.perf_counter() - started
+    deterministic = {
+        "returned_jobs": count,
+        "client_calls": client.get_jobs.await_count,
+        "fallback_calls": client.get_prompt_queue.await_count,
+        "visible_job_lines": sum(
+            line.startswith("- ") for line in rendered.splitlines()
+        ),
+        "summary_chars": len(rendered),
+        "digest": _canonical_digest(rendered),
+    }
+    return deterministic, elapsed
+
+
+class TestR218PerformanceBaseline(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.policy = _load_policy()
+
+    def test_policy_schema_is_bounded_advisory_and_content_free(self):
+        policy = self.policy
+        self.assertEqual(policy["schema_version"], 1)
+        self.assertEqual(policy["timing"]["enforcement"], "advisory_only")
+        self.assertEqual(policy["timing"]["samples"], 2)
+        ids = [item["id"] for item in policy["workloads"]]
+        self.assertEqual(set(ids), EXPECTED_WORKLOAD_IDS)
+        self.assertEqual(len(ids), len(set(ids)))
+        serialized = json.dumps(policy, sort_keys=True).lower()
+        for forbidden in (
+            "latency_threshold",
+            "max_seconds",
+            "b:\\",
+            "/home/",
+            "prompt",
+            "token",
+            "secret",
+        ):
+            self.assertNotIn(forbidden, serialized)
+        for item in policy["workloads"]:
+            self.assertIsInstance(item["seed"], int)
+            self.assertGreater(item["seed"], 0)
+            self.assertTrue(item["owner"])
+            self.assertTrue(item["review_after"])
+
+    def test_policy_validator_rejects_schema_budget_and_privacy_drift(self):
+        cases = []
+        unknown = copy.deepcopy(self.policy)
+        unknown["unexpected"] = True
+        cases.append(unknown)
+        duplicate = copy.deepcopy(self.policy)
+        duplicate["workloads"][1]["id"] = duplicate["workloads"][0]["id"]
+        cases.append(duplicate)
+        timed_gate = copy.deepcopy(self.policy)
+        timed_gate["timing"]["max_seconds"] = 1
+        cases.append(timed_gate)
+        unsafe_size = copy.deepcopy(self.policy)
+        unsafe_size["workloads"][0]["input"]["history_records"] = 100_000
+        cases.append(unsafe_size)
+        unsafe_content = copy.deepcopy(self.policy)
+        unsafe_content["policy_id"] = "B:\\private\\scale"
+        cases.append(unsafe_content)
+        missing_counter = copy.deepcopy(self.policy)
+        del missing_counter["workloads"][0]["expected"]["exact"]["examined"]
+        cases.append(missing_counter)
+        unknown_budget = copy.deepcopy(self.policy)
+        unknown_budget["workloads"][1]["expected"]["unexpected"] = 1
+        cases.append(unknown_budget)
+
+        self.assertEqual(validate_performance_policy(self.policy), [])
+        for invalid in cases:
+            with self.subTest(invalid=invalid):
+                self.assertTrue(validate_performance_policy(invalid))
+
+    def test_backend_jobs_history_matches_deterministic_budgets(self):
+        workload = _workload(self.policy, "backend_jobs_history")
+        result, elapsed = _run_backend_probe(workload)
+        expected = workload["expected"]
+        self.assertEqual(
+            {key: result[key] for key in expected["exact"]}, expected["exact"]
+        )
+        self.assertLessEqual(result["payload_bytes"], expected["max_payload_bytes"])
+        self.assertEqual(result["digest"], expected["digest_sha256"])
+        self.assertGreaterEqual(elapsed, 0.0)
+
+    def test_connector_dispatch_matches_deterministic_budgets(self):
+        workload = _workload(self.policy, "connector_jobs_dispatch")
+        result, elapsed = _run_connector_probe(workload)
+        expected = workload["expected"]
+        self.assertEqual(
+            {key: result[key] for key in expected["exact"]}, expected["exact"]
+        )
+        self.assertLessEqual(result["summary_chars"], expected["max_summary_chars"])
+        self.assertEqual(result["digest"], expected["digest_sha256"])
+        self.assertGreaterEqual(elapsed, 0.0)
+
+    def test_repeated_runs_compare_deterministic_results_not_timing(self):
+        for workload_id, probe in (
+            ("backend_jobs_history", _run_backend_probe),
+            ("connector_jobs_dispatch", _run_connector_probe),
+        ):
+            workload = _workload(self.policy, workload_id)
+            first, first_elapsed = probe(workload)
+            second, second_elapsed = probe(workload)
+            self.assertEqual(first, second)
+            self.assertGreaterEqual(first_elapsed, 0.0)
+            self.assertGreaterEqual(second_elapsed, 0.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

web/tests/unit/performance_baseline.test.js

@@ -0,0 +1,100 @@
+import { createHash } from "node:crypto";
+import { readFileSync } from "node:fs";
+import { resolve } from "node:path";
+import { describe, expect, it } from "vitest";
+
+import {
+    extractHistoryImageRefs,
+    extractHistoryOutputRefs,
+} from "../../openclaw_asset_refs.js";
+
+
+const policy = JSON.parse(readFileSync(
+    resolve(process.cwd(), "tests/performance_baseline_policy.json"),
+    "utf8",
+));
+const workload = policy.workloads.find((item) => item.id === "frontend_history_outputs");
+
+function stableValue(value) {
+    if (Array.isArray(value)) {
+        return value.map(stableValue);
+    }
+    if (value && typeof value === "object") {
+        return Object.fromEntries(
+            Object.keys(value).sort().map((key) => [key, stableValue(value[key])]),
+        );
+    }
+    return value;
+}
+
+function canonicalDigest(value) {
+    return createHash("sha256")
+        .update(JSON.stringify(stableValue(value)))
+        .digest("hex");
+}
+
+function seededGenerator(seed) {
+    let state = seed >>> 0;
+    return () => {
+        state = (Math.imul(state, 1664525) + 1013904223) >>> 0;
+        return state;
+    };
+}
+
+function runFrontendProbe() {
+    const next = seededGenerator(workload.seed);
+    const outputs = {};
+    const { nodes, refs_per_node: refsPerNode } = workload.input;
+    for (let nodeIndex = 0; nodeIndex < nodes; nodeIndex += 1) {
+        outputs[String(nodeIndex)] = {
+            images: Array.from({ length: refsPerNode }, (_, refIndex) => ({
+                filename: `scale-${nodeIndex.toString().padStart(3, "0")}-${refIndex}.png`,
+                subfolder: `batch-${next() % 8}`,
+                type: (next() % 2) === 0 ? "output" : "temp",
+            })),
+        };
+    }
+
+    const started = performance.now();
+    const normalized = extractHistoryOutputRefs({ outputs });
+    const images = extractHistoryImageRefs({ outputs });
+    const elapsedMs = performance.now() - started;
+    const serialized = JSON.stringify(normalized);
+    return {
+        deterministic: {
+            input_refs: nodes * refsPerNode,
+            normalized_outputs: normalized.length,
+            image_outputs: images.length,
+            serialized_bytes: Buffer.byteLength(serialized, "utf8"),
+            digest: canonicalDigest(normalized),
+        },
+        elapsedMs,
+    };
+}
+
+describe("R218 deterministic frontend scale baseline", () => {
+    it("matches fixed cardinality, serialization, and digest budgets", () => {
+        expect(policy.schema_version).toBe(1);
+        expect(policy.timing.enforcement).toBe("advisory_only");
+        expect(workload).toBeTruthy();
+
+        const { deterministic, elapsedMs } = runFrontendProbe();
+        const expected = workload.expected;
+        for (const [key, value] of Object.entries(expected.exact)) {
+            expect(deterministic[key]).toBe(value);
+        }
+        expect(deterministic.serialized_bytes).toBeLessThanOrEqual(
+            expected.max_serialized_bytes,
+        );
+        expect(deterministic.digest).toBe(expected.digest_sha256);
+        expect(elapsedMs).toBeGreaterThanOrEqual(0);
+    });
+
+    it("compares repeated deterministic output without enforcing elapsed time", () => {
+        const first = runFrontendProbe();
+        const second = runFrontendProbe();
+        expect(first.deterministic).toEqual(second.deterministic);
+        expect(first.elapsedMs).toBeGreaterThanOrEqual(0);
+        expect(second.elapsedMs).toBeGreaterThanOrEqual(0);
+    });
+});