test: benchmark particle-only C FastPath CPU across R layouts

tools/particle_android_validation/benchmark_step119_cpu.py

@@ -0,0 +1,192 @@
+#!/usr/bin/env python3
+"""STEP-119 / RCRR CPU 对照。只启动已安装测试场景并读取真机CPU/Native计数，不编译/修改24 Gate。"""
+import argparse
+import json
+import statistics
+import subprocess
+import sys
+import time
+from datetime import datetime
+from pathlib import Path
+
+import verify_step113_fast_path as verifier
+
+BENCH_SCENARIOS = (
+    "RCRR_FULL_CONTROL_NO_TIDE",
+    "RCRR_C_NARROW_NO_TIDE",
+    "RCR_PARTICLE_ONLY_NO_TIDE",
+    "RCRRR_PARTICLE_ONLY_NO_TIDE",
+)
+
+
+def parse_proc_stat(raw: str):
+    """Linux /proc/PID/stat: comm may contain whitespace or ')'; indices are post-comm."""
+    close = raw.rfind(")")
+    if close <= 0 or not raw.startswith(raw.split(" ", 1)[0] + " ("):
+        raise ValueError("invalid /proc/PID/stat header")
+    rest = raw[close + 2:].split()
+    if len(rest) < 20:
+        raise ValueError("truncated /proc/PID/stat")
+    # Fields 14/15 utime/stime and field 22 starttime; rest starts at field 3.
+    return int(rest[11]) + int(rest[12]), int(rest[19])
+
+
+def collect_result(samples, cpu_start, cpu_end, clock_ticks, elapsed):
+    native_start, native_end = samples
+    fast_delta = native_end["fastRendered"] - native_start["fastRendered"]
+    full_delta = native_end["fullRendered"] - native_start["fullRendered"]
+    interval = (native_end["sampleElapsedMs"] - native_start["sampleElapsedMs"]) / 1000
+    if interval <= 0 or fast_delta < 0 or full_delta < 0:
+        raise ValueError("invalid Native window/counters")
+    cpu_sec = (cpu_end["ticks"] - cpu_start["ticks"]) / clock_ticks
+    if cpu_sec < 0:
+        raise ValueError("CPU counter reset")
+    fps = (fast_delta + full_delta) / interval
+    cpu_cores = cpu_sec / elapsed
+    return {
+        "cpuSeconds": cpu_sec,
+        "cpuCoreEquivalent": cpu_cores,
+        "cpuPercentOfSingleCore": 100 * cpu_cores,
+        "nativeWindowSeconds": interval,
+        "nativeWindowFullFrames": full_delta,
+        "nativeWindowFastFrames": fast_delta,
+        "nativeObservedFps": fps,
+        "fastFraction": fast_delta / (fast_delta + full_delta) if fast_delta + full_delta else None,
+        "estimatedCpuMsPerNativeFrame": 1000 * cpu_cores / fps if fps else None,
+        "caveat": ("CPU时间为整个App进程，含UI、日志和其他线程；"
+                   "帧数由邻近原生摘要窗口得出；CPU毫秒/帧为近似值，不是原生逐帧计时。"),
+    }
+
+
+def run(args):
+    adb = args.adb or str(Path.home() / "Library/Android/sdk/platform-tools/adb")
+    info, sha, head, _ = verifier.inspect(verifier.APK, adb, args.serial)
+    def device(*argv):
+        return verifier.call([adb, "-s", args.serial, *argv])
+    expected_full = args.scenario == "RCRR_FULL_CONTROL_NO_TIDE"
+    clock_ticks = int(device("shell", "getconf", "CLK_TCK"))
+    if clock_ticks <= 0:
+        raise RuntimeError("invalid CLOCK_TICKS")
+    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    out = verifier.ARTIFACTS / "step119-rcrr-cpu" / f"{stamp}-{args.scenario.lower()}"
+    out.mkdir(parents=True, exist_ok=False)
+    launch = device("shell", "am", "start", "-S", "-n", verifier.ACTIVITY,
+                    "--es", "step113_scenario", args.scenario)
+    (out / "launch.txt").write_text(launch + "\n", encoding="utf-8")
+    pid = None
+    for _ in range(12):
+        time.sleep(1)
+        pid_tokens = verifier.call([adb, "-s", args.serial, "shell", "pidof",
+                                    verifier.PACKAGE], check=False).split()
+        if pid_tokens:
+            pid = int(pid_tokens[0])
+            break
+    if pid is None:
+        raise RuntimeError("APP_PID_MISSING")
+    print(f"SCENARIO={args.scenario} PID={pid} SHA_MATCHED={sha} SOURCE={head}", flush=True)
+    time.sleep(args.warmup)
+    warm_log = device("logcat", "-d", "-v", "brief", "-s", "STEP113Gate:I", "*:S")
+    warm_samples = verifier.extract(warm_log, pid, args.scenario)
+    if not any(s.get("sceneReady") and not s.get("blockedReason") for s in warm_samples):
+        (out / "warmup_logcat.txt").write_text(warm_log, encoding="utf-8")
+        raise RuntimeError("scene not ready after warmup (saved log)")
+
+    def cpu_sample():
+        raw = device("shell", "sh", "-c", f"cat /proc/{pid}/stat; cat /proc/uptime")
+        lines = [line for line in raw.splitlines() if line.strip()]
+        if len(lines) != 2:
+            raise RuntimeError(f"unexpected /proc sample: {raw[:120]}")
+        ticks, born = parse_proc_stat(lines[0])
+        uptime = float(lines[1].split()[0])
+        return {"ticks": ticks, "startTimeTicks": born, "deviceElapsedMs": uptime * 1000,
+                "hostMonotonic": time.monotonic()}
+
+    observations = [cpu_sample()]
+    deadline = observations[0]["hostMonotonic"] + args.seconds
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            break
+        time.sleep(min(args.interval, remaining))
+        sample = cpu_sample()
+        if sample["startTimeTicks"] != observations[0]["startTimeTicks"]:
+            raise RuntimeError("process PID reused/restarted: cannot compare")
+        observations.append(sample)
+    first_cpu, last_cpu = observations[0], observations[-1]
+    elapsed = last_cpu["hostMonotonic"] - first_cpu["hostMonotonic"]
+    raw_log = device("logcat", "-d", "-v", "brief", "-s", "STEP113Gate:I", "*:S")
+    (out / "logcat.txt").write_text(raw_log + "\n", encoding="utf-8")
+    native = verifier.extract(raw_log, pid, args.scenario)
+    ready = [s for s in native if s.get("sceneReady") and not s.get("blockedReason")
+             and isinstance(s.get("sampleElapsedMs"), (float, int))]
+    def closest(device_time):
+        if not ready:
+            raise RuntimeError("no timestamped Native summary")
+        return min(ready, key=lambda s: abs(s["sampleElapsedMs"] - device_time))
+    a, b = closest(first_cpu["deviceElapsedMs"]), closest(last_cpu["deviceElapsedMs"])
+    offsets = [abs(a["sampleElapsedMs"] - first_cpu["deviceElapsedMs"]),
+               abs(b["sampleElapsedMs"] - last_cpu["deviceElapsedMs"])]
+    parity = all(verifier.scene_parity_valid(s, args.scenario) for s in (a, b))
+    same_lifecycle = (a.get("sceneGeneration"), a.get("lifecycle")) == (
+        b.get("sceneGeneration"), b.get("lifecycle"))
+    control_match = all(s.get("commands", {}).get("forceFullControl") == expected_full
+                        for s in (a, b))
+    metrics = collect_result((a, b), first_cpu, last_cpu, clock_ticks, elapsed)
+    loads = []
+    for old, new in zip(observations, observations[1:]):
+        dt = new["hostMonotonic"] - old["hostMonotonic"]
+        if dt > 0:
+            loads.append((new["ticks"] - old["ticks"]) / clock_ticks / dt)
+    metrics["medianCpuCoreEquivalentPerInterval"] = statistics.median(loads) if loads else None
+    status = (
+        "INVALID_PARITY_OR_LIFECYCLE" if not parity or not same_lifecycle or not control_match else
+        "SUMMARY_ALIGNMENT_LIMIT" if max(offsets) > 9000 else
+        "DECISION_MISMATCH" if b.get("decisionMismatch") != 0 else
+        "FULL_CONTROL_OBSERVED" if expected_full and metrics["nativeWindowFullFrames"] > 0
+            and metrics["nativeWindowFastFrames"] == 0 else
+        "FAST_DOMINANT" if not expected_full and
+            metrics["fastFraction"] is not None and metrics["fastFraction"] >= 0.95 else
+        "MIXED_OR_NO_FAST" if not expected_full else "NO_EXPECTED_FRAMES"
+    )
+    data = {
+        "schema": "step119-rcrr-cpu-benchmark-v1",
+        "status": status, "sourceCommit": head, "apkSha256": sha,
+        "device": args.serial, "pid": pid, "scenario": args.scenario,
+        "fullControl": expected_full, "durationSecondsRequested": args.seconds,
+        "durationSecondsObserved": elapsed, "warmupSeconds": args.warmup,
+        "clockTicksPerSecond": clock_ticks, "nativeCount": len(ready),
+        "nativeCpuBoundaryOffsetMs": offsets, "sameLifecycle": same_lifecycle,
+        "parityValid": parity, "controlMatched": control_match,
+        "nativeStart": a, "nativeEnd": b,
+        "cpuObservations": observations, "metrics": metrics,
+        "performanceQualification": ("进程级CPU测量仅可与同APK/相同Style相机负载的对照比较；"
+            "应取Full-Fast-Full或交叉顺序重复，排除热机、缓存热度和帧率不同造成的误判；"
+            "不能用此数据宣称GPU时间、省电或逐帧CPU精确耗时。"),
+    }
+    (out / "result.json").write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n",
+                                    encoding="utf-8")
+    print(json.dumps({"status": status, "scenario": args.scenario, "metrics": metrics,
+                      "evidence": str(out / "result.json")}, ensure_ascii=False), flush=True)
+    return 0 if status in ("FAST_DOMINANT", "FULL_CONTROL_OBSERVED") else 2
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--serial", default="192.168.200.139:5555")
+    parser.add_argument("--adb", default=None)
+    parser.add_argument("--scenario", choices=BENCH_SCENARIOS, required=True)
+    parser.add_argument("--seconds", type=int, default=90)
+    parser.add_argument("--warmup", type=int, default=35)
+    parser.add_argument("--interval", type=float, default=6.0)
+    args = parser.parse_args()
+    if args.seconds < 75 or args.warmup < 20 or args.interval < 1:
+        parser.error("观察至少75秒、Warmup至少20秒、采样间隔至少1秒")
+    try:
+        return run(args)
+    except (RuntimeError, OSError, ValueError, KeyError) as e:
+        print(f"STEP119_CPU_BENCH_BLOCKED: {e}", file=sys.stderr)
+        return 3
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tools/particle_android_validation/compare_step119_cpu.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""STEP-119 RCRR CPU效益对比；仅整合已采集真机JSON，不会重新启动设备/构建。"""
+import argparse
+import json
+import statistics
+from datetime import datetime
+from pathlib import Path
+
+
+def read(path):
+    data = json.loads(Path(path).read_text(encoding="utf-8"))
+    if data.get("schema") != "step119-rcrr-cpu-benchmark-v1":
+        raise ValueError(f"invalid schema: {path}")
+    return data
+
+
+def check_common(reports):
+    ref = reports[0]
+    for r in reports:
+        for key in ("sourceCommit", "apkSha256", "device"):
+            if r.get(key) != ref.get(key):
+                raise ValueError(f"cross-run {key} mismatch: {r['scenario']}")
+        for key in ("fingerprint", "actualLayers"):
+            if r["nativeStart"].get(key) != ref["nativeStart"].get(key):
+                raise ValueError(f"style/actual layers mismatch {key}: {r['scenario']}")
+        if r["nativeStart"].get("actualLayers") != 64:
+            raise ValueError("not 64-layer isolated Test App")
+        if r["nativeEnd"].get("lifecycle") != r["nativeStart"].get("lifecycle"):
+            raise ValueError("scene lifecycle changed")
+        if r.get("nativeCpuBoundaryOffsetMs") is None or max(
+                r["nativeCpuBoundaryOffsetMs"]) > 9000:
+            raise ValueError("CPU/Native summary not well-aligned")
+        if not r.get("parityValid") or not r.get("controlMatched"):
+            raise ValueError("parity/control mismatch")
+        if r["nativeEnd"].get("decisionMismatch") != 0:
+            raise ValueError("FastGate decisionMismatch")
+    return {"commit": ref["sourceCommit"], "apkSha256": ref["apkSha256"],
+            "device": ref["device"], "styleFingerprint": ref["nativeStart"]["fingerprint"]}
+
+
+def summarize(full_before, fast, full_after, others):
+    inputs = [full_before, fast, full_after, *others]
+    identity = check_common(inputs)
+    if full_before["status"] != "FULL_CONTROL_OBSERVED" or full_after[
+            "status"] != "FULL_CONTROL_OBSERVED":
+        raise ValueError("Full control run missing/invalid")
+    if full_before["scenario"] != "RCRR_FULL_CONTROL_NO_TIDE" or full_after[
+            "scenario"] != "RCRR_FULL_CONTROL_NO_TIDE":
+        raise ValueError("Full AB control is not the same RCRR scene")
+    if fast["scenario"] != "RCRR_C_NARROW_NO_TIDE" or fast["status"] != "FAST_DOMINANT":
+        raise ValueError("fast run has no accepted particle-only-C FastPath")
+    def value(r, key):
+        x = r["metrics"].get(key)
+        if not isinstance(x, (int, float)) or x <= 0:
+            raise ValueError(f"invalid {key} in {r['scenario']}")
+        return x
+    baseline_ms = statistics.median([
+        value(full_before, "estimatedCpuMsPerNativeFrame"),
+        value(full_after, "estimatedCpuMsPerNativeFrame"),
+    ])
+    drift = abs(value(full_before, "cpuCoreEquivalent") -
+                value(full_after, "cpuCoreEquivalent")) / max(
+                    value(full_before, "cpuCoreEquivalent"), 1e-9)
+    comparisons = []
+    for r in [fast, *others]:
+        if r["status"] != "FAST_DOMINANT":
+            comparisons.append({"scenario": r["scenario"], "status": "FAST_NOT_STABLE",
+                                "sourceStatus": r["status"]})
+            continue
+        ms = value(r, "estimatedCpuMsPerNativeFrame")
+        comparisons.append({
+            "scenario": r["scenario"],
+            "cpuMsPerNativeFrameEstimated": ms,
+            "cpuMsPerFrameSavingPercentVsFull": 100 * (1 - ms / baseline_ms),
+            "cpuCoreEquivalent": value(r, "cpuCoreEquivalent"),
+            "nativeFps": value(r, "nativeObservedFps"),
+            "fastFraction": r["metrics"]["fastFraction"],
+            "buildReuse": [
+                {"segment": s["segment"], "build": s["build"], "reuse": s["reuse"],
+                 "invalidation": s["invalidation"]}
+                for s in r["nativeEnd"].get("retained", [])
+            ],
+        })
+    return {
+        "schema": "step119-rcrr-cpu-comparison-v1", "identity": identity,
+        "fullControlCpuMsPerFrameEstimatedMedian": baseline_ms,
+        "fullControlCpuCoreEquivalent": [value(full_before, "cpuCoreEquivalent"),
+                                         value(full_after, "cpuCoreEquivalent")],
+        "fullControlFrameRate": [value(full_before, "nativeObservedFps"),
+                                 value(full_after, "nativeObservedFps")],
+        "fullBeforeAfterCpuCoreRelativeDrift": drift,
+        "comparisons": comparisons,
+        "conclusion": "CPU_DRIFT_REQUIRES_RETEST" if drift > .15 else
+                      "NUMERICAL_COMPARISON_ONLY_NEEDS_VISUAL_AND_THERMAL_VALIDATION",
+        "caveats": ("同源/同层序A-B-A的进程CPU时间近似对比；Full/Fast FPS不同时必须同时解读CPU/秒、CPU/帧和FPS。"
+                    "CPU指标不等于GPU耗时或电量，未通过人工视觉和失效矩阵不能据此认定正式性能PASS。"),
+    }
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--full-before", required=True)
+    ap.add_argument("--fast", required=True)
+    ap.add_argument("--full-after", required=True)
+    ap.add_argument("--rcr")
+    ap.add_argument("--rcrrr")
+    ap.add_argument("--output")
+    args = ap.parse_args()
+    others = [read(p) for p in (args.rcr, args.rcrrr) if p]
+    result = summarize(read(args.full_before), read(args.fast),
+                       read(args.full_after), others)
+    target = Path(args.output) if args.output else (
+        Path(__file__).resolve().parents[2] /
+        "artifacts/maplibre-particle/android-validation/step119-rcrr-cpu" /
+        f"compare-{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n",
+                      encoding="utf-8")
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    print(f"EVIDENCE={target}")
+
+
+if __name__ == "__main__":
+    main()

tools/particle_android_validation/test_step119_rcrr_scenarios.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""STEP-119 RCRR 验证器结构契约测试；合成样本不代表真机 GTest 或 GPU PASS。"""
+"""STEP-119：基于真实 Style 索引的粒子独占 C 分段验证器测试。仅结构测试，非 GPU。"""
 import importlib.util
 import unittest
 from pathlib import Path
@@ -10,72 +10,171 @@ verifier = importlib.util.module_from_spec(spec)
 spec.loader.exec_module(verifier)
 
 
-def sample(retained):
+def sample(ranges):
     return {
-        "planValid": True,
-        "basePreserved": False,
-        "tideLayerExcluded": True,
-        "originalBaseLayerCount": 63,
-        "effectiveBaseLayerCount": 62,
-        "excludedLayerId": "tide-spots",
-        "excludedLayerSource": "tide",
-        "excludedLayerCount": 1,
-        "actualLayers": 64,
+        "planValid": True, "basePreserved": False, "tideLayerExcluded": True,
+        "originalBaseLayerCount": 63, "effectiveBaseLayerCount": 62,
+        "excludedLayerId": "tide-spots", "excludedLayerSource": "tide",
+        "excludedLayerCount": 1, "actualLayers": 64,
         "commands": {"emptyBuiltinAnnotationLayerRemoved": True,
                      "annotationSourcePreserved": True},
         "sources": [{"sourceId": "navsea_delivery"},
                     {"sourceId": "rn-weather-fields"}],
-        "retained": retained,
+        "retained": [{"begin": start, "end": end} for start, end in ranges],
+        "segmentRanges": (
+            [{"strategy": "RETAINED", "begin": ranges[0][0], "end": ranges[0][1]},
+             {"strategy": "CONTINUOUS", "begin": ranges[0][1], "end": ranges[1][0]}]
+            + [{"strategy": "RETAINED", "begin": a, "end": b}
+               for a, b in ranges[1:]]
+            if len(ranges) >= 2 else []
+        ),
     }
 
 
-def regions(r0_end, r2_start):
-    return [
-        {"begin": 0, "end": r0_end},
-        {"begin": r2_start, "end": 63},
-        {"begin": 63, "end": 64},
-    ]
+SHAPES = {
+    # Stable WeatherColor is in R[0,32); only Particle at [32,33) is C.
+    "RCRR_C_NARROW_NO_TIDE": [(0, 32), (33, 63), (63, 64)],
+    "RCRR_FULL_CONTROL_NO_TIDE": [(0, 32), (33, 63), (63, 64)],
+    "RCR_PARTICLE_ONLY_NO_TIDE": [(0, 32), (33, 64)],
+    "RCRRR_PARTICLE_ONLY_NO_TIDE": [(0, 32), (33, 38), (38, 63), (63, 64)],
+    # Intentionally unsupported boundaries: keep as independent negative scenes.
+    "RCRR_C_WIDE_NO_TIDE": [(0, 2), (38, 63), (63, 64)],
+    "RCRR_R_DYNAMIC_NO_TIDE": [(0, 33), (38, 63), (63, 64)],
+}
 
 
 class Step119RcrrVerifierTests(unittest.TestCase):
-    def test_three_rcrr_shapes_have_distinct_expected_boundaries(self):
-        shapes = {
-            "RCRR_C_NARROW_NO_TIDE": regions(31, 33),
-            "RCRR_C_WIDE_NO_TIDE": regions(2, 38),
-            "RCRR_R_DYNAMIC_NO_TIDE": regions(33, 38),
+    def test_all_expected_shapes_pass_exact_validation(self):
+        for name, ranges in SHAPES.items():
+            with self.subTest(name=name):
+                self.assertTrue(verifier.scene_parity_valid(sample(ranges), name))
+
+    def test_incorrect_ranges_do_not_pass_other_layouts(self):
+        for name, ranges in SHAPES.items():
+            for other in SHAPES:
+                if other == name or {name, other} == {
+                        "RCRR_C_NARROW_NO_TIDE", "RCRR_FULL_CONTROL_NO_TIDE"}:
+                    continue
+                with self.subTest(name=name, other=other):
+                    self.assertFalse(verifier.scene_parity_valid(sample(ranges), other))
+
+    def test_missing_source_or_annotation_guard_fails_closed(self):
+        for name, ranges in SHAPES.items():
+            a = sample(ranges)
+            a["sources"].append({"sourceId": "tide"})
+            self.assertFalse(verifier.scene_parity_valid(a, name))
+            b = sample(ranges)
+            b["commands"]["emptyBuiltinAnnotationLayerRemoved"] = False
+            self.assertFalse(verifier.scene_parity_valid(b, name))
+
+    def test_invalid_last_extent_or_count(self):
+        for name, ranges in SHAPES.items():
+            bad = [*ranges]
+            bad[-1] = (bad[-1][0], 65)
+            self.assertFalse(verifier.scene_parity_valid(sample(bad), name))
+            self.assertFalse(verifier.scene_parity_valid(sample(ranges[:-1]), name))
+
+    def test_actual_c_segment_must_match_particle_only_boundary(self):
+        exact = sample(SHAPES["RCRR_C_NARROW_NO_TIDE"])
+        exact["segmentRanges"][1]["end"] = 38
+        self.assertFalse(verifier.scene_parity_valid(
+            exact, "RCRR_C_NARROW_NO_TIDE"))
+        missing = sample(SHAPES["RCRR_C_NARROW_NO_TIDE"])
+        missing.pop("segmentRanges")
+        self.assertFalse(verifier.scene_parity_valid(
+            missing, "RCRR_C_NARROW_NO_TIDE"))
+
+    def test_full_control_may_omit_cache_but_not_segment_plan(self):
+        control = sample(SHAPES["RCRR_FULL_CONTROL_NO_TIDE"])
+        control["retained"] = []
+        self.assertTrue(verifier.scene_parity_valid(
+            control, "RCRR_FULL_CONTROL_NO_TIDE"))
+        control["segmentRanges"][1]["begin"] = 31
+        self.assertFalse(verifier.scene_parity_valid(
+            control, "RCRR_FULL_CONTROL_NO_TIDE"))
+
+    def test_static_rc_no_tide_old_contract_unchanged(self):
+        old = sample([(0, 62)])
+        self.assertTrue(verifier.scene_parity_valid(old, "STATIC_RC_NO_TIDE"))
+        self.assertFalse(verifier.scene_parity_valid(old, "RCRR_C_NARROW_NO_TIDE"))
+
+    def test_legacy_original_63_layer_fails_if_changed(self):
+        original = {"planValid": True, "basePreserved": True,
+                    "tideLayerExcluded": False}
+        self.assertTrue(verifier.scene_parity_valid(original, "PARTIAL_RCRR"))
+        original["tideLayerExcluded"] = True
+        self.assertFalse(verifier.scene_parity_valid(original, "PARTIAL_RCRR"))
+
+
+class Step119ProcessCpuTests(unittest.TestCase):
+    def test_proc_stat_parses_cpu_and_stable_birth_with_parentheses(self):
+        spec_cpu = importlib.util.spec_from_file_location(
+            "benchmark", Path(__file__).with_name("benchmark_step119_cpu.py"))
+        # Direct loading tests the real no-third-party benchmark implementation.
+        import sys
+        parent = str(Path(__file__).parent)
+        if parent not in sys.path:
+            sys.path.insert(0, parent)
+        benchmark = importlib.util.module_from_spec(spec_cpu)
+        spec_cpu.loader.exec_module(benchmark)
+        parts = ["S"] + ["0"] * 19
+        parts[11] = "120"
+        parts[12] = "20"
+        parts[19] = "90000"
+        ticks, born = benchmark.parse_proc_stat(
+            "222 (android app (bench)) " + " ".join(parts))
+        self.assertEqual((ticks, born), (140, 90000))
+
+    def test_cpu_per_frame_is_normalized_by_actual_native_fps(self):
+        import sys
+        parent = str(Path(__file__).parent)
+        if parent not in sys.path:
+            sys.path.insert(0, parent)
+        import benchmark_step119_cpu as benchmark
+        a = {"fastRendered": 100, "fullRendered": 5, "sampleElapsedMs": 1000}
+        b = {"fastRendered": 200, "fullRendered": 5, "sampleElapsedMs": 61000}
+        result = benchmark.collect_result(
+            (a, b), {"ticks": 200}, {"ticks": 500}, 100, 60.0)
+        self.assertAlmostEqual(result["cpuSeconds"], 3.0)
+        self.assertAlmostEqual(result["cpuCoreEquivalent"], .05)
+        self.assertAlmostEqual(result["nativeObservedFps"], 100.0 / 60.0)
+        self.assertAlmostEqual(result["estimatedCpuMsPerNativeFrame"], 30.0)
+
+
+class Step119CpuComparisonTests(unittest.TestCase):
+    @staticmethod
+    def report(scenario, status, ms, cores, fps):
+        native = {"fingerprint": "test-pbf", "actualLayers": 64,
+                  "lifecycle": 7, "decisionMismatch": 0, "retained": []}
+        return {
+            "sourceCommit": "source-locked", "apkSha256": "sha-locked",
+            "device": "test-device", "scenario": scenario, "status": status,
+            "nativeStart": dict(native), "nativeEnd": dict(native),
+            "nativeCpuBoundaryOffsetMs": [100, 300],
+            "parityValid": True, "controlMatched": True,
+            "metrics": {"estimatedCpuMsPerNativeFrame": ms, "cpuCoreEquivalent": cores,
+                        "nativeObservedFps": fps, "fastFraction": .98},
         }
-        for scenario, retained in shapes.items():
-            with self.subTest(scenario=scenario):
-                self.assertTrue(verifier.scene_parity_valid(sample(retained), scenario))
-                for other in shapes:
-                    if other != scenario:
-                        self.assertFalse(verifier.scene_parity_valid(sample(retained), other))
 
-    def test_missing_source_or_annotation_policy_never_passes(self):
-        for scenario, retained in [
-            ("RCRR_C_NARROW_NO_TIDE", regions(31, 33)),
-            ("RCRR_C_WIDE_NO_TIDE", regions(2, 38)),
-            ("RCRR_R_DYNAMIC_NO_TIDE", regions(33, 38)),
-        ]:
-            with self.subTest(scenario=scenario):
-                a = sample(retained)
-                a["sources"].append({"sourceId": "tide"})
-                self.assertFalse(verifier.scene_parity_valid(a, scenario))
-                b = sample(retained)
-                b["commands"]["emptyBuiltinAnnotationLayerRemoved"] = False
-                self.assertFalse(verifier.scene_parity_valid(b, scenario))
+    def test_valid_aba_cpu_saving_report(self):
+        import compare_step119_cpu as compare
+        a = self.report("RCRR_FULL_CONTROL_NO_TIDE", "FULL_CONTROL_OBSERVED", 40, .50, 15)
+        b = self.report("RCRR_C_NARROW_NO_TIDE", "FAST_DOMINANT", 20, .42, 28)
+        c = self.report("RCRR_FULL_CONTROL_NO_TIDE", "FULL_CONTROL_OBSERVED", 42, .51, 14)
+        result = compare.summarize(a, b, c, [])
+        self.assertAlmostEqual(result["fullControlCpuMsPerFrameEstimatedMedian"], 41)
+        self.assertAlmostEqual(result["comparisons"][0]["cpuMsPerFrameSavingPercentVsFull"],
+                               100 * (1 - 20 / 41))
+        self.assertNotEqual(result["conclusion"], "CPU_DRIFT_REQUIRES_RETEST")
 
-    def test_wrong_final_extent_and_wrong_retained_count_rejected(self):
-        a = sample(regions(31, 33))
-        a["retained"][-1]["end"] = 65
-        self.assertFalse(verifier.scene_parity_valid(a, "RCRR_C_NARROW_NO_TIDE"))
-        b = sample(regions(31, 33)[:2])
-        self.assertFalse(verifier.scene_parity_valid(b, "RCRR_C_NARROW_NO_TIDE"))
-
-    def test_existing_static_no_tide_contract_preserved(self):
-        a = sample([{"begin": 0, "end": 62}])
-        self.assertTrue(verifier.scene_parity_valid(a, "STATIC_RC_NO_TIDE"))
-        self.assertFalse(verifier.scene_parity_valid(a, "RCRR_C_NARROW_NO_TIDE"))
+    def test_aba_source_mismatch_is_rejected(self):
+        import compare_step119_cpu as compare
+        a = self.report("RCRR_FULL_CONTROL_NO_TIDE", "FULL_CONTROL_OBSERVED", 40, .5, 15)
+        b = self.report("RCRR_C_NARROW_NO_TIDE", "FAST_DOMINANT", 20, .4, 28)
+        c = self.report("RCRR_FULL_CONTROL_NO_TIDE", "FULL_CONTROL_OBSERVED", 41, .5, 15)
+        b["apkSha256"] = "different-apk"
+        with self.assertRaises(ValueError):
+            compare.summarize(a, b, c, [])
 
 
 if __name__ == "__main__":

tools/particle_android_validation/verify_step113_fast_path.py

@@ -19,9 +19,12 @@ BUILD_INFO = ARTIFACTS / "build-info.txt"
 PACKAGE = "org.maplibre.android.testapp"
 ACTIVITY = f"{PACKAGE}/{PACKAGE}.activity.particle.ParticleCoreLayerValidationActivity"
 SCENARIOS = ("STATIC_RC", "PBF63", "PARTICLE_OFF", "PARTIAL_RCRR", "SOURCE_CHANGE", "STATIC_RC_NO_TIDE",
-             "RCRR_C_NARROW_NO_TIDE", "RCRR_C_WIDE_NO_TIDE", "RCRR_R_DYNAMIC_NO_TIDE")
+             "RCRR_C_NARROW_NO_TIDE", "RCRR_C_WIDE_NO_TIDE", "RCRR_R_DYNAMIC_NO_TIDE",
+             "RCR_PARTICLE_ONLY_NO_TIDE", "RCRRR_PARTICLE_ONLY_NO_TIDE", "RCRR_FULL_CONTROL_NO_TIDE")
 NO_TIDE_SCENARIOS = {"STATIC_RC_NO_TIDE", "RCRR_C_NARROW_NO_TIDE",
-                     "RCRR_C_WIDE_NO_TIDE", "RCRR_R_DYNAMIC_NO_TIDE"}
+                     "RCRR_C_WIDE_NO_TIDE", "RCRR_R_DYNAMIC_NO_TIDE",
+                     "RCR_PARTICLE_ONLY_NO_TIDE", "RCRRR_PARTICLE_ONLY_NO_TIDE",
+                     "RCRR_FULL_CONTROL_NO_TIDE"}
 LOG_ENTRY = re.compile(r"^[A-Z]/STEP113Gate\(\s*(\d+)\):\s*(\{.+\})\s*$")
 
 
@@ -112,19 +115,36 @@ def extract(logcat, pid, expected):
 def rcrr_segments_valid(sample, scenario):
     """Verify exact R/C boundaries in the new isolated, 64-layer PBF Test App cases."""
     retained = sample.get("retained", [])
-    if len(retained) != 3:
+    ranges = [(entry.get("begin"), entry.get("end")) for entry in retained]
+    expected = {
+        "RCRR_C_NARROW_NO_TIDE": [(0, 32), (33, 63), (63, 64)],
+        "RCRR_FULL_CONTROL_NO_TIDE": [(0, 32), (33, 63), (63, 64)],
+        "RCR_PARTICLE_ONLY_NO_TIDE": [(0, 32), (33, 64)],
+        "RCRRR_PARTICLE_ONLY_NO_TIDE": [(0, 32), (33, 38), (38, 63), (63, 64)],
+        "RCRR_C_WIDE_NO_TIDE": [(0, 2), (38, 63), (63, 64)],
+        "RCRR_R_DYNAMIC_NO_TIDE": [(0, 33), (38, 63), (63, 64)],
+    }
+    expected_r = expected.get(scenario)
+    if expected_r is None:
         return False
-    r0, r2, r3 = retained
-    if (r0.get("begin") != 0 or r2.get("end") != r3.get("begin")
-            or r3.get("end") != 64 or r3.get("begin") != 63):
+    # The actual Native segment ranges must prove that the single C consists
+    # only of Particle for the normal CPU candidates. An R-only list cannot
+    # prove the C type/range; fail closed if the Native list is missing.
+    boundaries = sample.get("segmentRanges")
+    if not isinstance(boundaries, list):
         return False
-    if scenario == "RCRR_C_NARROW_NO_TIDE":
-        return r0.get("end") == 31 and r2.get("begin") == 33
-    if scenario == "RCRR_C_WIDE_NO_TIDE":
-        return r0.get("end") == 2 and r2.get("begin") == 38
-    if scenario == "RCRR_R_DYNAMIC_NO_TIDE":
-        return r0.get("end") == 33 and r2.get("begin") == 38
-    return False
+    expected_segments = [{"strategy": "RETAINED", "begin": expected_r[0][0],
+                          "end": expected_r[0][1]},
+                         {"strategy": "CONTINUOUS", "begin": expected_r[0][1],
+                          "end": expected_r[1][0]}]
+    expected_segments.extend({"strategy": "RETAINED", "begin": a, "end": b}
+                             for a, b in expected_r[1:])
+    if boundaries != expected_segments:
+        return False
+    # Cache-disabled Full control may omit cache diagnostics but must still
+    # supply the exact actual R/C segment list above.
+    return (not ranges or ranges == expected_r) if scenario == (
+        "RCRR_FULL_CONTROL_NO_TIDE") else ranges == expected_r
 
 
 def scene_parity_valid(sample, scenario):