fix(telemetry): derive workload gate bounds from the bucket above the baseline

The gate could not catch a regression on any sub-millisecond span.
compare_to_baseline.py requires both the percentage and the absolute bound to
breach, and every span shared one flat absolute bound of 10 ms (15 ms for p99)
calibrated for a 5-25 ms band the spans do not occupy. Against the baseline
captured on 2026-08-24, where 18 of the 28 quantiles gated at the time sat
below 1 ms, that bound sat 1.15x to 2000x above the metric it guarded, so the
AND never fired: a 100x regression injected into span.ledger.store.p95 reported
0 regressions and exit 0. Injecting a 10x regression into each key in turn was
caught on only 5 of 28.

Give every gated key its own absolute bound, equal to the distance from its
baseline to hi_next, the edge above the top of the bucket the baseline sits in.
The trip point is then exactly hi_next, so the gate fires only once the reading
clears the bucket above the baseline's own. That is the property a multiple of
the enclosing bucket width cannot provide: after the quantile crosses hi, the
interpolation happens across the next bucket, which on this ladder is up to
eight times wider, so no multiple of the enclosing width bounds the excursion.
Measured with a model-free reachability test, a single bucket crossing can
produce a false regression on 2 of 25 keys under the old flat bound and 0 of 25
under this rule. The smallest catchable regression is 2.02x to 9.43x per key.

The job queue bound had the same shape of problem on three of its four keys
(42x, 47x, 220x before). Defaults now sit at each ladder floor, leaving the
percentage bound operative for a metric that somehow reaches them.

Drop span.ledger.store from the gated surface. Its captured quantiles were
0.005, 0.0095 and 0.0099 ms, which is the ladder's 0.01 ms floor times the
quantile: every sample lands under 10 us, so the reported value does not move
even if each store slows from 2 us to 9 us. No bound can gate it. Presence is
still asserted by expected_spans.json and the integration test, and the rate is
still on the ledger-operations dashboard.

Add check_regression_bounds.py, wired into the same workflow step as the bucket
parity check. It fails when a bound is not the one its own baseline implies,
when a gated key has no override, when the baseline and metric surface disagree,
when the percentage bound would become operative, and when a baseline carries
the ladder floor signature. This gate has now broken three times through the
same drift between ladder, baseline and bounds, so documentation alone is not
enough.

compare_to_baseline.py is unchanged: its existing per-metric override mechanism
already expresses all of this.

A missing, unreadable or malformed input makes that check exit 1 naming the
input, rather than reporting success without having checked anything; only a
placeholder baseline, the documented bootstrap state, still exits 0. Its own
tests cover both halves of that contract plus one case per rule, and run in the
workflow before the check so a broken rule reads as a broken rule.

.github/scripts/telemetry/check_regression_bounds.py

@@ -0,0 +1,284 @@
+#!/usr/bin/env python3
+"""Assert every workload-gate absolute bound is the one its own baseline implies.
+
+The regression gate in ``docker/telemetry/workload`` fails CI when a span or
+job-queue quantile grows. Whether it *can* fail is decided by
+``regression-thresholds.json``, and that file's numbers are derived from
+``baselines/baseline-timings.json`` plus the two histogram ladders. Nothing
+tied the three together, and the gate has now been broken three times by the
+same class of drift:
+
+  1. the microsecond ladder's floor moved 100us -> 1us, voiding every
+     job_queue baseline captured before it;
+  2. the spanmetrics ladder's floor moved 1ms -> 0.01ms, voiding every
+     sub-millisecond span baseline captured before it;
+  3. the absolute bounds stayed calibrated for a 5-25ms band the spans had
+     left, so a 100x regression on ``span.ledger.store.p95`` reported zero
+     regressions and exit 0.
+
+Each time the gate stayed green, which is indistinguishable from a passing
+build. Documentation did not prevent recurrence, so this is a check.
+
+The rule it enforces is the one recorded in ``regression-thresholds.json``
+under ``_absolute_bound_derivation``: for a baseline sitting in the half-open
+bucket ``(lo, hi]`` of its ladder, with ``hi_next`` the next edge above ``hi``,
+
+    max_abs_increase_* == hi_next - baseline
+
+so the gate trips only when the reading clears the bucket *above* the
+baseline's own. Five rules are checked:
+
+  A  the baseline's key set equals the surface ``regression-metrics.json``
+     declares (a stale key left behind reads as covered but never gates);
+  B  every gated key has a per-metric override, not a fallback default;
+  C  each absolute bound equals ``hi_next - baseline``;
+  D  each percentage bound stays below ``100 * bound / baseline``, so the
+     absolute bound remains the operative half of the ``AND`` -- the span
+     ladder's 2s/3s/4s edges are only 1.25x-1.5x apart, where this silently
+     stops being true;
+  E  no baseline carries the ladder-floor signature ``quantile x first_edge``,
+     which means every sample landed in the first bucket and the number is
+     interpolation arithmetic rather than a latency.
+
+A PLACEHOLDER baseline -- ``"placeholder": true`` or an empty ``metrics``
+object -- exits 0, because that is the documented bootstrap state and CI has to
+stay green while a baseline is being recaptured. A missing, unreadable or
+malformed input is a different thing and exits 1: a check that reports success
+without having checked anything is the same green-build-that-is-not failure this
+script exists to prevent, so renaming or deleting one of its inputs must not
+silence it.
+
+Exit 0 when every rule holds, 1 with per-key detail otherwise.
+"""
+
+import json
+import re
+import sys
+from pathlib import Path
+
+WORKLOAD = Path("docker/telemetry/workload")
+BASELINE = WORKLOAD / "baselines/baseline-timings.json"
+THRESHOLDS = WORKLOAD / "regression-thresholds.json"
+METRICS = WORKLOAD / "regression-metrics.json"
+COLLECTOR = Path("docker/telemetry/otel-collector-config.yaml")
+HEADER = Path("include/xrpl/telemetry/HistogramBuckets.h")
+
+UNIT_TO_MS = {"ms": 1.0, "s": 1000.0}
+# A bound may differ from the derived value only by double round-tripping.
+REL_TOLERANCE = 1e-12
+
+
+def read_text_or_exit(path):
+    """Read a required text input, or exit 1 naming the input that failed."""
+    try:
+        return path.read_text()
+    except OSError as exc:
+        sys.exit(f"{path}: required input could not be read -- {exc}")
+
+
+def read_json_or_exit(path):
+    """Read and parse a required JSON input, or exit 1 naming what failed."""
+    try:
+        return json.loads(read_text_or_exit(path))
+    except json.JSONDecodeError as exc:
+        sys.exit(f"{path}: required input is not valid JSON -- {exc}")
+
+
+def span_edges_ms():
+    """Parse the spanmetrics bucket list, normalising each edge to milliseconds."""
+    match = re.search(r"buckets:\s*\[(.*?)\]", read_text_or_exit(COLLECTOR), re.S)
+    if not match:
+        sys.exit(f"{COLLECTOR}: no 'buckets:' list found")
+    edges = []
+    for raw in match.group(1).split(","):
+        token = raw.strip()
+        if not token:
+            continue
+        parsed = re.fullmatch(r"([0-9.]+)(ms|s)", token)
+        if not parsed:
+            sys.exit(f"{COLLECTOR}: cannot parse bucket edge {token!r}")
+        edges.append(float(parsed.group(1)) * UNIT_TO_MS[parsed.group(2)])
+    return edges
+
+
+def microsecond_edges():
+    """Parse kMicrosecondBuckets out of the header that owns every ladder."""
+    match = re.search(r"kMicrosecondBuckets\{(.*?)\};", read_text_or_exit(HEADER), re.S)
+    if not match:
+        sys.exit(f"{HEADER}: kMicrosecondBuckets not found")
+    return [
+        float(token.strip().replace("'", ""))
+        for token in match.group(1).split(",")
+        if token.strip()
+    ]
+
+
+def declared_keys(metrics_cfg):
+    """Rebuild the flat key set regression-metrics.json declares.
+
+    Deliberately reimplemented rather than imported from ``prom_queries.py``,
+    which pulls in aiohttp; CI telemetry checks stay dependency-free. The key
+    format is fixed by that file's own ``_key_format`` field.
+    """
+    keys = set()
+    spans = metrics_cfg.get("spans", {})
+    for name in spans.get("names", []):
+        for quantile in spans.get("_quantiles", []):
+            keys.add(f"span.{name}.p{_quantile_label(quantile)}")
+    jobs = metrics_cfg.get("job_queue", {})
+    for name in jobs.get("names", []):
+        for phase in jobs.get("_phases", []):
+            for quantile in jobs.get("_quantiles", []):
+                keys.add(f"job.{name}.{phase}.p{_quantile_label(quantile)}")
+    return keys
+
+
+def _quantile_label(quantile):
+    """0.95 -> '95', 0.5 -> '50', matching capture_timings.py's key format."""
+    return f"{quantile * 100:g}".replace(".", "")
+
+
+def brackets(value, edges):
+    """Return ``(lo, hi, hi_next)`` for the bucket ``(lo, hi]`` holding value."""
+    padded = [0.0] + list(edges)
+    for i in range(1, len(padded)):
+        if value <= padded[i]:
+            hi_next = padded[i + 1] if i + 1 < len(padded) else None
+            return padded[i - 1], padded[i], hi_next
+    return None, None, None
+
+
+def resolve_override(key, thresholds):
+    """Return the override rule for a key, or None if it falls back to defaults."""
+    group, quantile = key.rsplit(".", 1)
+    return thresholds.get("overrides", {}).get(group, {}).get(quantile)
+
+
+def check_key(key, entry, thresholds, ladders):
+    """Apply rules B, C, D and E to one gated key. Returns a list of failures."""
+    value, unit = entry.get("value"), entry.get("unit", "")
+    edges = ladders.get(unit)
+    if value is None or edges is None:
+        return [f"{key}: baseline has no value, or unknown unit {unit!r}"]
+
+    failures = []
+    first_edge = edges[0]
+    quantile = int(key.rsplit(".p", 1)[1]) / 100.0
+    if abs(value - quantile * first_edge) <= 1e-9 * first_edge:
+        failures.append(
+            f"{key}: baseline {value!r} equals quantile {quantile:g} x the ladder "
+            f"floor {first_edge:g}{unit}, so every sample landed in the first "
+            f"bucket and this is bucket arithmetic, not a latency. No absolute "
+            f"bound can gate it -- add a finer ladder edge or drop the metric "
+            f"from {METRICS} (rule E)"
+        )
+        return failures
+
+    _, _, hi_next = brackets(value, edges)
+    if hi_next is None:
+        return [
+            f"{key}: baseline {value!r}{unit} sits in or above the ladder's top "
+            f"bucket, so there is no hi_next to derive a bound from -- extend the "
+            f"ladder (rule C)"
+        ]
+
+    rule = resolve_override(key, thresholds)
+    if rule is None:
+        failures.append(
+            f"{key}: no per-metric override, so it falls back to the defaults and "
+            f"gates on the percentage bound alone. Add an override with "
+            f"max_abs_increase = {hi_next - value!r} (rule B)"
+        )
+        return failures
+
+    bound = rule.get("max_abs_increase_ms", rule.get("max_abs_increase_us"))
+    expected = hi_next - value
+    if bound is None or abs(bound - expected) > REL_TOLERANCE * expected:
+        failures.append(
+            f"{key}: absolute bound is {bound!r}, expected {expected!r} "
+            f"(hi_next {hi_next:g} - baseline {value!r}) (rule C)"
+        )
+
+    pct = rule.get("max_pct_increase")
+    if pct is None:
+        failures.append(f"{key}: no max_pct_increase, so the metric never gates")
+    elif bound is not None and pct >= 100.0 * bound / value:
+        failures.append(
+            f"{key}: max_pct_increase {pct:g}% is at or above the absolute bound's "
+            f"{100.0 * bound / value:.1f}% of baseline, so the percentage bound "
+            f"becomes the operative one and the bucket guarantee is lost. Lower it "
+            f"or document the metric as percentage-gated (rule D)"
+        )
+    return failures
+
+
+def main():
+    missing = [
+        p for p in (BASELINE, THRESHOLDS, METRICS, COLLECTOR, HEADER) if not p.exists()
+    ]
+    if missing:
+        print("Cannot check workload regression bounds.", file=sys.stderr)
+        for path in missing:
+            print(f"  {path}: required input is absent", file=sys.stderr)
+        print(
+            "\nA missing input is not a reason to pass. Deleting or renaming one of\n"
+            "these would otherwise leave the gate reporting success without having\n"
+            "checked a single bound -- the failure this script exists to prevent. If\n"
+            "the workload harness has genuinely moved, update the paths here.",
+            file=sys.stderr,
+        )
+        return 1
+
+    baseline = read_json_or_exit(BASELINE)
+    thresholds = read_json_or_exit(THRESHOLDS)
+    metrics_cfg = read_json_or_exit(METRICS)
+
+    if baseline.get("placeholder") is True or not baseline.get("metrics"):
+        print("OK: baseline is a placeholder, bounds cannot be derived yet")
+        return 0
+
+    ladders = {"ms": span_edges_ms(), "us": microsecond_edges()}
+    gated = baseline["metrics"]
+    failures = []
+
+    declared = declared_keys(metrics_cfg)
+    for key in sorted(set(gated) - declared):
+        failures.append(
+            f"{key}: in the baseline but not declared by {METRICS}, so it is "
+            f"reported every run and can never gate -- remove it (rule A)"
+        )
+    for key in sorted(declared - set(gated)):
+        failures.append(
+            f"{key}: declared by {METRICS} but absent from the baseline, so it "
+            f"never gates -- capture a baseline for it (rule A)"
+        )
+
+    for key in sorted(gated):
+        if key in declared:
+            failures.extend(check_key(key, gated[key], thresholds, ladders))
+
+    if not failures:
+        print(
+            f"OK: {len(gated)} gated key(s); every absolute bound equals "
+            f"hi_next - baseline, every key has an override, and the absolute "
+            f"bound is the operative half of the AND for all of them"
+        )
+        return 0
+
+    print(
+        "Workload regression bounds are not derived from the baseline.", file=sys.stderr
+    )
+    for failure in failures:
+        print(f"  {failure}", file=sys.stderr)
+    print(
+        f"\nThe rule is recorded in {THRESHOLDS} under _absolute_bound_derivation:\n"
+        "a bound is hi_next - baseline, where hi_next is the edge above the top of\n"
+        "the bucket holding the baseline. Refreshing a baseline therefore obliges\n"
+        "you to re-derive its bound; see baselines/README.md.",
+        file=sys.stderr,
+    )
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.github/scripts/telemetry/test_check_regression_bounds.py

@@ -0,0 +1,226 @@
+#!/usr/bin/env python3
+"""Tests for check_regression_bounds.py.
+
+The checker reads five files by path relative to the working directory, so each
+test assembles a scratch tree holding copies of the real inputs, mutates one
+thing, and runs the checker as a subprocess there. Testing the real entry point
+is deliberate: the contract under test is the exit code CI reads, and an
+in-process call would not exercise it.
+
+Two groups:
+
+* the input-handling contract -- a placeholder baseline must PASS because that
+  is the documented bootstrap state, while a missing, unreadable or malformed
+  input must FAIL. A checker that returns success without having checked
+  anything is the failure this whole gate exists to prevent;
+* one case per rule (A to E), so a rule that stops flagging is caught.
+
+stdlib unittest only; the repo installs no third-party runner for CI.
+"""
+
+import json
+import os
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+CHECKER = SCRIPT_DIR / "check_regression_bounds.py"
+REPO = SCRIPT_DIR.parents[2]
+
+WORKLOAD = "docker/telemetry/workload"
+BASELINE = f"{WORKLOAD}/baselines/baseline-timings.json"
+THRESHOLDS = f"{WORKLOAD}/regression-thresholds.json"
+METRICS = f"{WORKLOAD}/regression-metrics.json"
+COLLECTOR = "docker/telemetry/otel-collector-config.yaml"
+HEADER = "include/xrpl/telemetry/HistogramBuckets.h"
+INPUTS = (BASELINE, THRESHOLDS, METRICS, COLLECTOR, HEADER)
+
+
+class CheckerCase(unittest.TestCase):
+    """Base class giving each test an isolated copy of the checker's inputs."""
+
+    def setUp(self):
+        self.tree = Path(tempfile.mkdtemp())
+        self.addCleanup(self._cleanup)
+        for rel in INPUTS:
+            dest = self.tree / rel
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy(REPO / rel, dest)
+        script = self.tree / ".github/scripts/telemetry/check_regression_bounds.py"
+        script.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy(CHECKER, script)
+
+    def _cleanup(self):
+        for path in self.tree.rglob("*"):
+            if path.is_file():
+                path.chmod(stat.S_IRUSR | stat.S_IWUSR)
+        shutil.rmtree(self.tree, ignore_errors=True)
+
+    def run_checker(self):
+        """Run the checker in the scratch tree, returning (code, stdout+stderr)."""
+        proc = subprocess.run(
+            [sys.executable, ".github/scripts/telemetry/check_regression_bounds.py"],
+            cwd=self.tree,
+            capture_output=True,
+            text=True,
+        )
+        return proc.returncode, proc.stdout + proc.stderr
+
+    def edit_json(self, rel, mutate):
+        """Load a scratch input, hand it to mutate(), write it back."""
+        path = self.tree / rel
+        data = json.loads(path.read_text())
+        mutate(data)
+        path.write_text(json.dumps(data, indent=2))
+
+
+class TestInputHandling(CheckerCase):
+    """A placeholder passes; a missing or broken input must not."""
+
+    def test_unmodified_tree_passes(self):
+        code, out = self.run_checker()
+        self.assertEqual(code, 0, out)
+        self.assertIn("gated key(s)", out)
+
+    def test_placeholder_flag_passes(self):
+        self.edit_json(BASELINE, lambda d: d.update(placeholder=True))
+        code, out = self.run_checker()
+        self.assertEqual(code, 0, out)
+        self.assertIn("placeholder", out)
+
+    def test_empty_metrics_baseline_passes(self):
+        self.edit_json(BASELINE, lambda d: d.update(metrics={}))
+        code, out = self.run_checker()
+        self.assertEqual(code, 0, out)
+        self.assertIn("placeholder", out)
+
+    def test_missing_baseline_fails_naming_the_input(self):
+        (self.tree / BASELINE).unlink()
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("baseline-timings.json", out)
+
+    def test_missing_collector_config_fails_naming_the_input(self):
+        (self.tree / COLLECTOR).unlink()
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("otel-collector-config.yaml", out)
+
+    @unittest.skipIf(os.geteuid() == 0, "root ignores the read permission bit")
+    def test_unreadable_baseline_fails(self):
+        (self.tree / BASELINE).chmod(0)
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("baseline-timings.json", out)
+        self.assertIn("could not be read", out)
+        self.assertNotIn("Traceback", out)
+
+    def test_malformed_baseline_json_fails(self):
+        (self.tree / BASELINE).write_text("{ not json")
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("valid JSON", out)
+
+    def test_malformed_thresholds_json_fails(self):
+        (self.tree / THRESHOLDS).write_text("]")
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("valid JSON", out)
+
+
+class TestRules(CheckerCase):
+    """One case per rule, so a rule that stops flagging is caught."""
+
+    def test_rule_a_flags_baseline_key_not_declared(self):
+        self.edit_json(
+            BASELINE,
+            lambda d: d["metrics"].update(
+                {"span.rpc.process.p99": {"unit": "ms", "value": 9.0}}
+            ),
+        )
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("(rule A)", out)
+
+    def test_rule_a_flags_declared_key_without_baseline(self):
+        self.edit_json(METRICS, lambda d: d["spans"]["names"].append("consensus.round"))
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("(rule A)", out)
+
+    def test_rule_b_flags_missing_override(self):
+        self.edit_json(THRESHOLDS, lambda d: d["overrides"].pop("span.ledger.build"))
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("(rule B)", out)
+
+    def test_rule_c_flags_rounded_bound(self):
+        self.edit_json(
+            THRESHOLDS,
+            lambda d: d["overrides"]["span.tx.process"]["p99"].update(
+                max_abs_increase_ms=4.0055
+            ),
+        )
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("(rule C)", out)
+
+    def test_rule_c_accepts_bound_within_relative_tolerance(self):
+        """The tolerance is 1e-12 relative, not exact equality."""
+        exact = 4.005485184848892
+        self.edit_json(
+            THRESHOLDS,
+            lambda d: d["overrides"]["span.tx.process"]["p99"].update(
+                max_abs_increase_ms=exact * (1 + 5e-13)
+            ),
+        )
+        code, out = self.run_checker()
+        self.assertEqual(code, 0, out)
+
+    def test_rule_d_flags_percentage_bound_becoming_operative(self):
+        self.edit_json(
+            THRESHOLDS,
+            lambda d: d["overrides"]["span.tx.apply"]["p99"].update(
+                max_pct_increase=150.0
+            ),
+        )
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("(rule D)", out)
+
+    def test_rule_e_flags_ladder_floor_signature(self):
+        """ledger.store's quantiles were the ladder floor times the quantile."""
+        store = {"p50": 0.005, "p95": 0.0095, "p99": 0.0099}
+        self.edit_json(METRICS, lambda d: d["spans"]["names"].append("ledger.store"))
+        self.edit_json(
+            BASELINE,
+            lambda d: d["metrics"].update(
+                {
+                    f"span.ledger.store.{q}": {"unit": "ms", "value": v}
+                    for q, v in store.items()
+                }
+            ),
+        )
+        self.edit_json(
+            THRESHOLDS,
+            lambda d: d["overrides"].update(
+                {
+                    "span.ledger.store": {
+                        q: {"max_pct_increase": 50.0, "max_abs_increase_ms": 0.05 - v}
+                        for q, v in store.items()
+                    }
+                }
+            ),
+        )
+        code, out = self.run_checker()
+        self.assertEqual(code, 1, out)
+        self.assertIn("(rule E)", out)
+
+
+if __name__ == "__main__":
+    unittest.main()