test(scanner): require bounded retry window evidence (#7517)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

docs/operations/scanner-benchmark-runbook.md

@@ -159,6 +159,14 @@ delay counts are both retained so an operator can reject unrelated or
 process-lifetime counter contamination. Correct repair oracles and the existing
 regression limits still apply in every case.
 
+The `running-heal` build comparison also records a `w11` section for the bounded
+retry-window evidence. `status=observed` requires same-window healthy-page
+latency improvement, reduced heal lock-wait p99, bounded candidate RSS growth,
+and a candidate attempt-cost value. `rss_regression` means latency and lock-wait
+improved but RSS exceeded the allowed growth limit; `no_measured_benefit` means
+attempt-cost evidence exists without the full W11 benefit; `pending` means the
+attempt-cost evidence needed for the comparison is missing.
+
 For P2, `measure.convergence` contains booleans `writes_stopped`,
 `last_mutation_observed`, `first_complete_publication`; numeric
 `last_mutation_time`, `last_mutation_observed_time`, `writes_stopped_time`, `window_start`, `window_end`,
@@ -220,7 +228,10 @@ non-passing report.
 Measured passing reports must also retain the W10/W11 foreground-pressure,
 heal-lock-wait, and heal-attempt-cost fields emitted by the ABBA evaluator. If
 those fields are removed, empty, malformed, or length-mismatched, the quiet
-summary fails closed instead of treating the report as performance evidence.
+summary fails closed instead of treating the report as performance evidence. For
+`running-heal` build comparisons, the summary additionally requires the emitted
+W11 section to be `observed` and to retain the RSS-growth, lock-wait,
+healthy-page-latency, and candidate attempt-cost values.
 
 They cover the complete 120-cell schedule, data isolation, missing builds and
 oracles, zero samples/requests, swallowed request errors, offered-load drift,

scripts/scanner_abba.py

@@ -30,6 +30,7 @@ METRICS = (
 )
 REPEATABILITY_LIMIT = Decimal("0.05")
 P2_WORK_MULTIPLE_LIMIT = Decimal("1.2")
+W11_RSS_GROWTH_LIMIT = Decimal("0.05")
 RELEASE_PROFILE_ARTIFACTS = (
     "allocation-profile",
     "flamegraph",
@@ -465,6 +466,48 @@ def running_heal_pacing(group, baseline, candidate, p99, throughput, noisy):
     }
 
 
+def bounded_retry_window(group, baseline, candidate, p99, throughput, noisy, candidate_attempt_costs):
+    if group[0]["scenario"] != "running-heal" or group[0]["comparison"] != "build":
+        return {"status": "not_applicable"}
+
+    rss_growth = relative_change_or_none(candidate["rss_bytes"], baseline["rss_bytes"], "rss_bytes")
+    lock_wait_change = relative_change_or_none(
+        candidate["heal_lock_wait_p99_ms"], baseline["heal_lock_wait_p99_ms"], "heal lock wait p99",
+    )
+    latency_improved = p99 < 0 or throughput > 0
+    lock_wait_improved = lock_wait_change is not None and lock_wait_change < 0
+    rss_within_limit = rss_growth is not None and rss_growth <= W11_RSS_GROWTH_LIMIT
+    attempt_cost_available = bool(candidate_attempt_costs)
+    status = (
+        "inconclusive"
+        if noisy
+        else "observed"
+        if latency_improved and lock_wait_improved and rss_within_limit and attempt_cost_available
+        else "rss_regression"
+        if latency_improved and lock_wait_improved and not rss_within_limit
+        else "no_measured_benefit"
+        if attempt_cost_available
+        else "pending"
+    )
+    return {
+        "status": status,
+        "rss_growth_limit": float(W11_RSS_GROWTH_LIMIT),
+        "rss_growth": None if rss_growth is None else float(rss_growth),
+        "rss_within_limit": rss_within_limit,
+        "baseline_rss_bytes": float(baseline["rss_bytes"]),
+        "candidate_rss_bytes": float(candidate["rss_bytes"]),
+        "baseline_heal_lock_wait_p99_ms": float(baseline["heal_lock_wait_p99_ms"]),
+        "candidate_heal_lock_wait_p99_ms": float(candidate["heal_lock_wait_p99_ms"]),
+        "heal_lock_wait_p99_change": None if lock_wait_change is None else float(lock_wait_change),
+        "healthy_page_latency_observed": latency_improved,
+        "foreground_p99_change": float(p99),
+        "foreground_throughput_change": float(throughput),
+        "candidate_attempt_cost_per_healed_object": (
+            None if not candidate_attempt_costs else float(max(candidate_attempt_costs))
+        ),
+    }
+
+
 def convergence(result):
     window = result.get("convergence")
     if not window or window.get("writes_stopped") is not True or window.get("last_mutation_observed") is not True or window.get("first_complete_publication") is not True:
@@ -525,6 +568,7 @@ def evaluate(cells):
             value for cell, value in zip(group, attempt_costs) if cell["leg"].startswith("B") and value is not None
         ]
         w10 = running_heal_pacing(group, a, b, p99, throughput, noise)
+        w11 = bounded_retry_window(group, a, b, p99, throughput, noise, candidate_attempt_costs)
         inconclusive |= noise or p2_pending
         if not noise and not passed:
             failed = True
@@ -541,6 +585,7 @@ def evaluate(cells):
                                 "candidate_vs_baseline": scanner_cache_cost_change(b, a),
                             },
                             "w10": w10,
+                            "w11": w11,
                             "w10_w11": {
                                 "foreground_pressure_high_sample_ratios": [
                                     float(pressure_high_ratio(cell["result"]["metrics"])) for cell in group

scripts/summarize_scanner_heal_perf.py

@@ -151,6 +151,30 @@ def require_measured_comparison_evidence(comparison: dict[str, Any], index: int)
             f"comparison {index} heal_lock_hold_p95_ms must be measured")
     require(len(start_p95) == len(duplicate_tasks) == len(lock_hold),
             f"comparison {index} W09 evidence length mismatch")
+    w11 = comparison.get("w11")
+    require(isinstance(w11, dict), f"comparison {index} missing W11 evidence")
+    status = w11.get("status")
+    require(status in {"observed", "no_measured_benefit", "rss_regression", "pending", "inconclusive", "not_applicable"},
+            f"comparison {index} invalid W11 evidence status")
+    if comparison.get("scenario") == "running-heal" and comparison.get("comparison") == "build":
+        require(status == "observed", f"comparison {index} W11 bounded retry evidence was not observed")
+        for key in (
+            "rss_growth_limit",
+            "rss_growth",
+            "baseline_rss_bytes",
+            "candidate_rss_bytes",
+            "baseline_heal_lock_wait_p99_ms",
+            "candidate_heal_lock_wait_p99_ms",
+            "heal_lock_wait_p99_change",
+            "foreground_p99_change",
+            "foreground_throughput_change",
+            "candidate_attempt_cost_per_healed_object",
+        ):
+            value = maybe_number(w11.get(key), f"comparison {index} W11 {key}")
+            require(value is not None, f"comparison {index} W11 {key} is required")
+        require(w11.get("rss_within_limit") is True, f"comparison {index} W11 RSS growth is outside limit")
+        require(w11.get("healthy_page_latency_observed") is True,
+                f"comparison {index} W11 healthy-page latency benefit is required")
 
 
 def require_complete_abba_matrix(manifest: dict[str, Any], report: dict[str, Any], comparisons: list[dict[str, Any]]) -> None:
@@ -206,6 +230,7 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
     start_p95_values: list[Decimal | None] = []
     duplicate_task_values: list[Decimal | None] = []
     lock_hold_values: list[Decimal | None] = []
+    w11_rows = []
     for index, comparison in enumerate(comparisons):
         require(isinstance(comparison, dict), f"comparison {index} must be an object")
         state = comparison.get("status")
@@ -238,6 +263,18 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
                 duplicate_task_values.append(maybe_number(value, "heal_duplicate_task_count"))
             for value in w09.get("heal_lock_hold_p95_ms", []):
                 lock_hold_values.append(maybe_number(value, "heal_lock_hold_p95_ms"))
+        w11 = comparison.get("w11")
+        if isinstance(w11, dict) and comparison.get("scenario") == "running-heal" and comparison.get("comparison") == "build":
+            w11_rows.append({
+                "round": comparison.get("round"),
+                "status": w11.get("status"),
+                "rss_growth": w11.get("rss_growth"),
+                "rss_growth_limit": w11.get("rss_growth_limit"),
+                "heal_lock_wait_p99_change": w11.get("heal_lock_wait_p99_change"),
+                "foreground_p99_change": w11.get("foreground_p99_change"),
+                "foreground_throughput_change": w11.get("foreground_throughput_change"),
+                "candidate_attempt_cost_per_healed_object": w11.get("candidate_attempt_cost_per_healed_object"),
+            })
 
     measured = report.get("evidence") == "measured"
     passed = report_state in PASS_STATES and performance_state in PASS_STATES and measured
@@ -280,6 +317,7 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
         "w09_duplicate_task_count": None if max_decimal(duplicate_task_values) is None else float(max_decimal(duplicate_task_values)),
         "w09_worst_lock_hold_p95_ms": None if max_decimal(lock_hold_values) is None else float(max_decimal(lock_hold_values)),
         "p1_reductions": p1_rows,
+        "w11_running_heal_build": w11_rows,
         "provenance": {
             "abba_dir": str(abba_dir.resolve()),
             "manifest_sha256": digest(manifest_path),
@@ -383,6 +421,9 @@ def markdown(summary: dict[str, Any]) -> str:
         f"- w09_duplicate_task_count: {abba['w09_duplicate_task_count'] if abba['w09_duplicate_task_count'] is not None else 'pending'}",
         f"- w09_worst_lock_hold_p95_ms: {abba['w09_worst_lock_hold_p95_ms'] if abba['w09_worst_lock_hold_p95_ms'] is not None else 'pending'}",
     ]
+    if abba.get("w11_running_heal_build"):
+        w11_statuses = ",".join(str(row.get("status")) for row in abba["w11_running_heal_build"])
+        lines.append(f"- w11_running_heal_build_statuses: {w11_statuses}")
     if abba.get("completed_cells") is not None:
         lines.append(f"- completed_cells: {abba['completed_cells']}")
     if abba.get("error"):

scripts/test_scanner_abba.py

@@ -114,6 +114,10 @@ def fake_adapter():
         elif fault == "pacing-benefit" and request["scenario"] == "running-heal" \
                 and request["comparison"] == "build" and request["leg"].startswith("B"):
             result["metrics"].update(p99_ms=9, heal_mainline_throttle_delayed=5)
+        elif fault == "w11-benefit" and request["scenario"] == "running-heal" \
+                and request["comparison"] == "build" and request["leg"].startswith("B"):
+            result["metrics"].update(p99_ms=9, throughput_ops=102, rss_bytes=10,
+                                     heal_lock_wait_p99_ms=5)
         elif fault == "pacing-pending" and request["scenario"] == "running-heal" \
                 and request["comparison"] == "build" and request["leg"].startswith("B"):
             result["metrics"]["heal_mainline_throttle_delayed"] = 0
@@ -378,6 +382,12 @@ class ScannerAbbaTest(unittest.TestCase):
                     "heal_lock_hold_p95_ms": [10, 10, 10, 10],
                 },
             )
+            if comparison["scenario"] == "running-heal" and comparison["comparison"] == "build":
+                self.assertEqual(comparison["w11"]["status"], "no_measured_benefit")
+                self.assertTrue(comparison["w11"]["rss_within_limit"])
+                self.assertFalse(comparison["w11"]["healthy_page_latency_observed"])
+            else:
+                self.assertEqual(comparison["w11"], {"status": "not_applicable"})
 
     def test_fail_closed_adapter_and_data_errors(self):
         for fault in ("measure-exit", "oracle-exit", "missing-oracle", "oracle-mismatch", "zero-samples",
@@ -427,6 +437,16 @@ class ScannerAbbaTest(unittest.TestCase):
                 build = next(comparison for comparison in comparisons if comparison["comparison"] == "build")
                 self.assertEqual(build["w10"]["status"], expected)
 
+    def test_running_heal_w11_status_requires_latency_lock_and_bounded_rss(self):
+        with patch.object(harness, "SCENARIOS", ("running-heal",)):
+            self.assertEqual(self.run_harness("w11-benefit"), 0)
+        comparisons = harness.read_json(self.root / "out/report.json")["comparisons"]
+        build = next(comparison for comparison in comparisons if comparison["comparison"] == "build")
+        self.assertEqual(build["w11"]["status"], "observed")
+        self.assertLess(build["w11"]["foreground_p99_change"], 0)
+        self.assertLess(build["w11"]["heal_lock_wait_p99_change"], 0)
+        self.assertTrue(build["w11"]["rss_within_limit"])
+
     def test_missing_first_publication_is_inconclusive(self):
         with patch.object(harness, "SCENARIOS", ("cold-hot",)):
             self.assertEqual(self.run_harness("no-publication"), 3)

scripts/test_summarize_scanner_heal_perf.py

@@ -104,6 +104,7 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
                 "heal_duplicate_task_count": [0, 0, 0, 0],
                 "heal_lock_hold_p95_ms": [7.0, 6.0, 6.5, 7.5],
             },
+            "w11": {"status": "not_applicable"},
         }
         self.report = {
             "status": "pass",
@@ -121,6 +122,22 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
                 for round_id in range(1, 4):
                     row = copy.deepcopy(self.comparison)
                     row.update(scenario=scenario, comparison=comparison, round=round_id)
+                    if scenario == "running-heal" and comparison == "build":
+                        row["w11"] = {
+                            "status": "observed",
+                            "rss_growth_limit": 0.05,
+                            "rss_growth": 0.01,
+                            "rss_within_limit": True,
+                            "baseline_rss_bytes": 1000000.0,
+                            "candidate_rss_bytes": 1010000.0,
+                            "baseline_heal_lock_wait_p99_ms": 12.0,
+                            "candidate_heal_lock_wait_p99_ms": 8.0,
+                            "heal_lock_wait_p99_change": -0.33,
+                            "healthy_page_latency_observed": True,
+                            "foreground_p99_change": -0.02,
+                            "foreground_throughput_change": 0.01,
+                            "candidate_attempt_cost_per_healed_object": 1.3,
+                        }
                     comparisons.append(row)
         return comparisons
 
@@ -157,6 +174,11 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
         self.assertEqual(result["abba"]["provenance"]["manifest_sha256"], sha(self.abba / "manifest.json"))
         self.assertEqual(result["abba"]["w09_duplicate_task_count"], 0.0)
         self.assertEqual(result["abba"]["w09_worst_heal_start_p95_ms"], 43.0)
+        self.assertEqual(
+            [row["status"] for row in result["abba"]["w11_running_heal_build"]],
+            ["observed", "observed", "observed"],
+        )
+        self.assertIn("w11_running_heal_build_statuses: observed,observed,observed", summary.markdown(result))
         self.assertEqual(result["cache_cost"]["max_save_body_amplification"], 2.0)
 
     def test_synthetic_report_fails_as_performance_conclusion(self):
@@ -247,7 +269,7 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
             summary.build_summary(args)
 
     def test_passing_abba_report_requires_w10_w11_evidence(self):
-        for fault in ("missing", "pressure", "lock", "attempt", "length", "range"):
+        for fault in ("missing", "pressure", "lock", "attempt", "length", "range", "w11-missing", "w11-pending"):
             with self.subTest(fault=fault):
                 self.setUp()
                 target = self.report["comparisons"][0]
@@ -261,6 +283,18 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
                     del target["w10_w11"]["attempt_cost_per_healed_object"]
                 elif fault == "length":
                     target["w10_w11"]["attempt_cost_per_healed_object"] = [None]
+                elif fault == "w11-missing":
+                    running_heal = next(
+                        comparison for comparison in self.report["comparisons"]
+                        if comparison["scenario"] == "running-heal" and comparison["comparison"] == "build"
+                    )
+                    del running_heal["w11"]
+                elif fault == "w11-pending":
+                    running_heal = next(
+                        comparison for comparison in self.report["comparisons"]
+                        if comparison["scenario"] == "running-heal" and comparison["comparison"] == "build"
+                    )
+                    running_heal["w11"]["status"] = "pending"
                 else:
                     target["w10_w11"]["foreground_pressure_high_sample_ratios"] = [1.5, 0.0, 0.0, 0.0]
                 self.write_inputs()
@@ -271,7 +305,7 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
                     "json_out": None,
                     "markdown_out": None,
                 })
-                with self.assertRaisesRegex(ValueError, "W10/W11|performance evidence|length mismatch|above maximum"):
+                with self.assertRaisesRegex(ValueError, "W10/W11|W11|performance evidence|length mismatch|above maximum"):
                     summary.build_summary(args)
 
     def test_passing_abba_report_requires_w09_evidence(self):