[Deferred] test: validate resource measurements and reject malformed CPU evidence

.github/workflows/mvp-ci.yml

@@ -55,6 +55,10 @@ jobs:
           python3 -m unittest discover -s scripts/ci -p 'test_*.py'
           node scripts/ci/test_planner_merge.js
 
+      - name: Test resource measurement normalization
+        working-directory: ASAPQuery-backend
+        run: python3 -m unittest discover -s tools/empirical-bench -p 'test_*.py'
+
       - name: Check workspace formatting
         working-directory: ASAPQuery-backend
         # `--all` also follows this workspace's sibling path dependencies.

control_plane/tests/resource_measurements.rs

@@ -0,0 +1,40 @@
+//! Work multiplicities and units, using explicit synthetic resource coefficients.
+#[path = "support/erp_contract.rs"]
+mod fixture;
+use control_plane::physical::erp::{ErpAccuracyMode, ErpParameterDecision};
+use planner_types::post_asap::{SketchAlgorithm, SketchParams};
+
+/// CPU seconds and byte-seconds scale independently; error statistics are unchanged.
+#[test]
+fn erp_resource_composition_uses_workload_multiplicities() {
+    for (updates, reads, merges, retention) in [(1000., 100., 0., 60.), (2000., 400., 5., 120.)] {
+        let mut policy = fixture::policy(ErpAccuracyMode::Empirical);
+        policy.expected_updates = updates;
+        policy.expected_queries = reads;
+        policy.expected_merges = merges;
+        policy.retention_seconds = retention;
+        let resources = &policy.artifact.records[0].resources;
+        let expected = policy.cpu_weight
+            * (updates * resources.update_cpu_seconds
+                + reads * resources.query_cpu_seconds
+                + merges * resources.merge_cpu_seconds)
+            + policy.byte_second_weight * resources.memory_bytes * retention;
+        let ErpParameterDecision::Empirical {
+            estimated_cost,
+            record_id,
+            ..
+        } = policy.select(
+            SketchAlgorithm::Cms,
+            0.01,
+            SketchParams::Cms {
+                width: 4096,
+                depth: 5,
+            },
+        )
+        else {
+            panic!("applicable ERP must be priced")
+        };
+        assert!((estimated_cost - expected).abs() < 1e-12);
+        assert_eq!(record_id, policy.artifact.records[0].id);
+    }
+}

docs/design_docs/resource-measurement-validation.md

@@ -0,0 +1,19 @@
+# Resource measurement validation
+
+This PR validates the existing sketch-bench export path and ERP workload
+composition. The Python tests distinguish process CPU from wall time, preserve
+paired repetitions, reject missing/invalid inputs and normalize milliseconds to
+nanoseconds per operation. The Rust test checks update/read/merge multiplicities
+and retained byte-seconds while retaining the ERP record identity.
+
+The regression first failed because `cpu_batch` silently truncated unmatched
+user/system samples and accepted a negative raw component when their sum was
+positive. `cpu_per_op` could also turn infinite work into a zero price. Both now
+reject these inputs. Missing CPU remains unknown, not zero.
+
+The tests use synthetic coefficients to validate arithmetic. They do not claim
+those coefficients are real measurements. Real resource evidence must retain raw
+paired samples, implementation/build/machine identity and measurement scope.
+The existing `tools/empirical-bench/run.py` records real benchmark provenance;
+its offline microbenchmarks do not independently establish deployment selection
+quality. Shared-work costing tests remain in `physical/workload_cost.rs`.

tools/empirical-bench/run.py

@@ -48,9 +48,11 @@ def cpu_per_op(row, op):
         raise ValueError("unaligned CPU/rate/time samples")
     values = []
     for r, elapsed, user, system in zip(*samples):
+        if any(not math.isfinite(x) or x < 0 for x in (r, elapsed, user, system)):
+            raise ValueError("invalid raw CPU/rate/time measurement")
         work = r * elapsed / 1000
-        if work <= 0:
-            raise ValueError("zero benchmark work")
+        if not math.isfinite(work) or work <= 0:
+            raise ValueError("invalid benchmark work")
         values.append((user + system) * 1_000_000 / work)
     result = measurement(values)
     result["method"] = "mean of paired run (process user+system CPU ns)/(throughput*wall seconds); per " + ("binary merge" if op == "merge" else "input item" if op == "insert" else "point-frequency key lookup")
@@ -61,8 +63,12 @@ def cpu_batch(row, op):
     cpu = row.get(op + "_cpu_time_ms")
     if not cpu:
         return None
-    return measurement([(u + s) * 1_000_000 for u, s in
-                        zip(cpu["user_ms"]["samples"], cpu["sys_ms"]["samples"])])
+    user, system = cpu["user_ms"]["samples"], cpu["sys_ms"]["samples"]
+    if len(user) != len(system):
+        raise ValueError("unaligned user/system CPU samples")
+    if any(not math.isfinite(x) or x < 0 for x in user + system):
+        raise ValueError("invalid raw CPU measurement")
+    return measurement([(u + s) * 1_000_000 for u, s in zip(user, system)])
 
 
 def export(raw, manifest, operation_reports, memory_rows, resource_rows=None):

tools/empirical-bench/test_resource_measurements.py

@@ -0,0 +1,52 @@
+"""Resource normalization contracts; values are synthetic, never calibration data."""
+import math
+import unittest
+from run import cpu_batch, cpu_per_op
+
+
+class ResourceMeasurementTests(unittest.TestCase):
+    def test_batch_rejects_unpaired_cpu_samples(self):
+        for user, system in [([1, 2], [3]), ([1], [2, 3]), ([], [1])]:
+            with self.subTest(user=user, system=system):
+                with self.assertRaises(ValueError):
+                    cpu_batch({"query_cpu_time_ms": {
+                        "user_ms": {"samples": user}, "sys_ms": {"samples": system}}}, "query")
+
+    def test_batch_and_per_op_reject_invalid_raw_measurements(self):
+        for invalid in [-1, math.inf, math.nan]:
+            with self.subTest(invalid=invalid):
+                with self.assertRaises(ValueError):
+                    cpu_batch({"query_cpu_time_ms": {
+                        "user_ms": {"samples": [invalid]}, "sys_ms": {"samples": [10]}}}, "query")
+        for field in ["rate", "elapsed", "user", "system"]:
+            for invalid in [-1, math.inf, math.nan]:
+                values = dict(rate=100, elapsed=1000, user=1, system=1)
+                values[field] = invalid
+                row = {"query_throughput_items_per_sec": {"samples": [values["rate"]]},
+                       "query_wall_time_ms": {"samples": [values["elapsed"]]},
+                       "query_cpu_time_ms": {"user_ms": {"samples": [values["user"]]},
+                                             "sys_ms": {"samples": [values["system"]]}}}
+                with self.subTest(field=field, invalid=invalid):
+                    with self.assertRaises(ValueError):
+                        cpu_per_op(row, "query")
+
+    def test_cpu_conversion_preserves_repetitions_and_units(self):
+        row = {"query_cpu_time_ms": {"user_ms": {"samples": [2, 4]},
+                                     "sys_ms": {"samples": [1, 2]}}}
+        batch = cpu_batch(row, "query")
+        self.assertEqual(batch["value"], 4_500_000)
+        self.assertEqual(batch["samples"], 2)
+        row.update(query_throughput_items_per_sec={"samples": [300, 300]},
+                   query_wall_time_ms={"samples": [1000, 2000]})
+        per_op = cpu_per_op(row, "query")
+        self.assertEqual(per_op["value"], 10_000)
+        self.assertEqual(per_op["stddev"], 0)
+        self.assertEqual(per_op["samples"], 2)
+
+    def test_missing_resource_dimension_stays_unknown(self):
+        self.assertIsNone(cpu_batch({}, "query"))
+        self.assertIsNone(cpu_per_op({}, "query"))
+
+
+if __name__ == "__main__":
+    unittest.main()