test(perf): measure the offload heartbeat against its own idle baseline, not a 7.5ms clock (#15221)

autobot-backend/code_intelligence/shared/process_offload_test.py

@@ -22,16 +22,34 @@
 against a 5 ms request. That per-wakeup tax, compounded across every await in a
 request, is how a 25 ms endpoint became a 12 s one. Everything else here is a
 property of the plumbing — this is the property the issue is about.
+
+HOW THE TAX IS MEASURED (#15221). Both numbers above were once also checked
+against a wall-clock ceiling — ``in_process < 0.005 * 1.5``. A 7.5 ms ceiling on
+a 5 ms heartbeat leaves 1.5x headroom on a machine that is also the CI runner,
+and it failed at 50.6 ms on a PR whose entire diff was two session-ownership test
+files. Its own message read "something is still contending for this process",
+which on a shared runner is both the regression it guards and the ordinary
+weather: the two are indistinguishable through an absolute clock.
+
+The replacement is the heartbeat's delay against ITS OWN idle baseline — the
+same 5 ms sleep, on the same loop, in windows either side of the scan. #15207's
+CPU work unit does not transfer here: timed during the scan it is starved by the
+very contention under test and the signal cancels, and timed outside it swung
+1.43-3.27 ms across six samples on this host, a wider spread than the 2x signal.
+The idle baseline held to within 1.4% over the same samples. See
+``assert_within_baseline_ratio``.
 """
 
 from __future__ import annotations
 
 import asyncio
 import time
 from pathlib import Path
+from typing import NamedTuple
 
 import pytest
 
+from autobot_shared.perf_work_budget import assert_within_baseline_ratio, recording_work_units
 from code_intelligence.shared import process_offload
 from code_intelligence.shared.process_offload import (
     run_directory_scan,
@@ -92,6 +110,33 @@ def analyze_directory(self, directory=None):
         raise RuntimeError("scan exploded in the child")
 
 
+#: The heartbeat's requested interval. CPython's default GIL switch interval is
+#: also 5 ms, and that is the tax the in-thread path adds to each wakeup.
+_HEARTBEAT_SECONDS = 0.005
+
+#: Quiet loop either side of the scan, whose ticks form the baseline. Two windows
+#: rather than one because they BRACKET the measurement in time: a slow drift in
+#: runner load then lands on baseline and measurement alike instead of on one.
+_IDLE_BASELINE_WINDOW_SECONDS = 0.3
+
+#: Fewest ticks a window may contribute before its median describes anything.
+_MIN_TICKS_PER_WINDOW = 10
+
+#: How far the heartbeat may drift above its own idle baseline while a scan runs
+#: in another process. Measured over three runs on the #15221 host: 1.009, 1.016
+#: and 1.023 offloaded, against 2.006, 2.015 and 2.024 with the scan forced
+#: in-thread. 1.5 clears the highest offloaded reading by 47% and still fails the
+#: in-thread regression by a third. Ratchet DOWN as runs report lower — never up.
+_TICK_BUDGET_VS_IDLE = 1.5
+
+
+@pytest.fixture(autouse=True)
+def _record_perf(record_property):
+    """#15055: send the measurement to the junit XML, green runs included."""
+    with recording_work_units(record_property):
+        yield
+
+
 @pytest.fixture(autouse=True)
 async def _fresh_pool():
     """Each test gets a pool it can break without affecting the next."""
@@ -149,34 +194,90 @@ def __init__(self, **kwargs):
     assert captured["exclude_patterns"] == ["venv"]
 
 
-async def _median_tick_delay_during_scan(force_thread: bool) -> float:
-    """Median time a 5 ms heartbeat actually takes while a GIL-bound scan runs."""
+class _TickLatency(NamedTuple):
+    """A heartbeat's median wakeup delay, with a scan in flight and without."""
+
+    busy_ms: float
+    idle_ms: float
+    busy_ticks: int
+    idle_ticks: int
+    pool_unavailable: str | None
+
+
+def _median_ms(samples: list[float]) -> float:
+    ordered = sorted(samples)
+    return ordered[len(ordered) // 2] * 1000.0
+
+
+async def _tick_latency_during_scan(force_thread: bool) -> _TickLatency:
+    """Time a 5 ms heartbeat while a GIL-bound scan runs, and on the quiet loop.
+
+    The idle windows are the yardstick: the same quantity as the measurement,
+    on the same loop in the same process, taken a fraction of a second either
+    side of it, differing only in whether a scan was in flight. Runner load
+    raises both terms and the ratio falls back toward 1.0, so a busy machine
+    makes the caller's budget more conservative rather than more likely to fire —
+    provided the load is comparable across the three windows, which is why the
+    baseline brackets the measurement instead of preceding it. A burst confined
+    to the scan window alone still moves the ratio up; see
+    ``assert_within_baseline_ratio`` for what that premise does and does not buy.
+    """
     await shutdown_scan_pool()
     process_offload._pool_unavailable_reason = "forced for measurement" if force_thread else None
 
     delays: list[float] = []
-    stop = False
+    beating = True
 
     async def heartbeat():
-        while not stop:
+        while beating:
             started = time.monotonic()
-            await asyncio.sleep(0.005)
+            await asyncio.sleep(_HEARTBEAT_SECONDS)
             delays.append(time.monotonic() - started)
 
+    def drain() -> list[float]:
+        taken = list(delays)
+        delays.clear()
+        return taken
+
     beat = asyncio.create_task(heartbeat())
     await asyncio.sleep(0.05)
     await run_directory_scan(_BurnCPU(project_root="/tmp"))  # warm the pool
-    delays.clear()
 
-    await run_directory_scan(_BurnCPU(project_root="/tmp"))
+    drain()
+    await asyncio.sleep(_IDLE_BASELINE_WINDOW_SECONDS)
+    idle = drain()
 
-    stop = True
+    await run_directory_scan(_BurnCPU(project_root="/tmp"))
+    busy = drain()
+
+    # Read before the reset below, or the caller's skip guard can never fire: on
+    # a host that forbids process creation the scan quietly runs in-thread and
+    # the comparison would report a regression that is really an environment.
+    unavailable = None if force_thread else process_offload._pool_unavailable_reason
+
+    # Resampled with no settling gap, deliberately: this scan leaves nothing
+    # behind that could still be taxing the loop, so the first tick after it is
+    # already an idle one. A workload whose effects outlive it — pool teardown,
+    # a GC pause — would need a gap here, or that tax would land in the BASELINE
+    # and shrink the ratio, weakening the very detection this exists for.
+    await asyncio.sleep(_IDLE_BASELINE_WINDOW_SECONDS)
+    idle += drain()
+
+    beating = False
     beat.cancel()
     await shutdown_scan_pool()
     process_offload._pool_unavailable_reason = None
 
-    assert delays, "the heartbeat never ran — measurement is meaningless"
-    return sorted(delays)[len(delays) // 2]
+    assert len(busy) >= _MIN_TICKS_PER_WINDOW, (
+        f"the heartbeat produced {len(busy)} ticks while the scan ran — fewer than "
+        f"{_MIN_TICKS_PER_WINDOW}, so there is no median to compare and the "
+        "measurement is meaningless"
+    )
+    assert len(idle) >= _MIN_TICKS_PER_WINDOW, (
+        f"the heartbeat produced {len(idle)} ticks on the idle loop — fewer than "
+        f"{_MIN_TICKS_PER_WINDOW}, so there is no baseline to compare against"
+    )
+    return _TickLatency(_median_ms(busy), _median_ms(idle), len(busy), len(idle), unavailable)
 
 
 async def test_a_scan_in_a_process_does_not_delay_the_event_loop():
@@ -191,25 +292,51 @@ async def test_a_scan_in_a_process_does_not_delay_the_event_loop():
     Compared within the test rather than against a fixed threshold, so a loaded
     CI runner shifts both numbers together instead of flaking.
     """
-    in_thread = await _median_tick_delay_during_scan(force_thread=True)
-    in_process = await _median_tick_delay_during_scan(force_thread=False)
+    in_thread = await _tick_latency_during_scan(force_thread=True)
+    in_process = await _tick_latency_during_scan(force_thread=False)
 
-    if process_offload._pool_unavailable_reason:
-        pytest.skip(f"process pool unavailable here: {process_offload._pool_unavailable_reason}")
+    if in_process.pool_unavailable:
+        pytest.skip(f"process pool unavailable here: {in_process.pool_unavailable}")
 
     # Measured ratio is ~0.5; 0.75 leaves headroom while still failing outright
     # if the scan ever stops being isolated.
-    assert in_process < in_thread * 0.75, (
-        f"a scan still delays the event loop: {in_process * 1000:.1f}ms median tick "
-        f"in-process vs {in_thread * 1000:.1f}ms in-thread — expected a clear improvement"
+    assert in_process.busy_ms < in_thread.busy_ms * 0.75, (
+        f"a scan still delays the event loop: {in_process.busy_ms:.1f}ms median tick "
+        f"in-process vs {in_thread.busy_ms:.1f}ms in-thread — expected a clear improvement"
     )
-    # And in absolute terms the loop should be barely taxed at all.
-    assert in_process < 0.005 * 1.5, (
-        f"a 5ms heartbeat took {in_process * 1000:.1f}ms median while a scan ran in "
-        "another process — something is still contending for this process"
+    # And the loop should be barely taxed at all — measured against the ticks it
+    # was serving moments earlier with nothing running, not against a clock.
+    assert_within_baseline_ratio(
+        in_process.busy_ms,
+        in_process.idle_ms,
+        _TICK_BUDGET_VS_IDLE,
+        f"5ms heartbeat during an offloaded scan ({in_process.busy_ticks} ticks " f"vs {in_process.idle_ticks} idle)",
     )
 
 
+async def test_the_responsiveness_budget_cannot_pass_without_a_measurement():
+    """A vacuity probe for the budget above (#15221).
+
+    The failure being guarded against is a budget that goes green because the
+    heartbeat never ran, the scan never started, or a refactor handed the
+    assertion a stub — the same way a wall-clock ceiling passes on an idle box
+    whether or not the code under it works. Every degenerate input must raise.
+    """
+    for measured_ms, baseline_ms, why in [
+        (0.0, 5.0, "a heartbeat that never ticked"),
+        (-1.0, 5.0, "a negative measurement"),
+        (None, 5.0, "no measurement at all"),
+        (5.0, 0.0, "a baseline that never ticked"),
+        (None, None, "neither term measured"),
+    ]:
+        with pytest.raises(AssertionError):
+            assert_within_baseline_ratio(measured_ms, baseline_ms, _TICK_BUDGET_VS_IDLE, why)
+
+    # Positive control, so the probe is not merely watching an assertion that
+    # can never pass: a real pair inside budget still goes green.
+    assert_within_baseline_ratio(5.1, 5.0, _TICK_BUDGET_VS_IDLE, "vacuity-probe control")
+
+
 async def test_run_isolated_forwards_args_and_kwargs():
     """The non-BaseCodeAnalyzer scanners take positional and keyword arguments."""
 

autobot_shared/perf_work_budget.py

@@ -136,3 +136,89 @@ def assert_within_work_budget(elapsed_ms: float, budget_units: float, what: str)
         f"{report}. This is a load-invariant ratio, not a wall-clock ceiling — "
         f"investigate the code, do not raise it."
     )
+
+
+def assert_within_baseline_ratio(
+    measured_ms: float,
+    baseline_ms: float,
+    budget_ratio: float,
+    what: str,
+) -> None:
+    """Assert a measurement stayed within `budget_ratio` of its OWN idle baseline.
+
+    The sibling of `assert_within_work_budget`, for quantities a CPU work unit
+    cannot normalise (#15221).
+
+    WHY THE WORK UNIT DOES NOT TRANSFER TO A CONTENTION MEASURE. `reference_workload()`
+    is a CPU-throughput yardstick. It normalises a numerator that is also CPU
+    throughput. It does not normalise event-loop wakeup latency, for two
+    independent reasons:
+
+    * Timed WHILE the operation under test runs, the yardstick is starved by
+      exactly the contention the assertion exists to detect, so numerator and
+      denominator inflate together and the signal cancels. A test that cannot
+      fail is not a test.
+    * Timed BEFORE or AFTER instead, it describes a different moment and a
+      different physical quantity. Measured on the #15221 host it spread
+      1.43ms-3.27ms across six consecutive samples — a 2.3x swing in the
+      DENOMINATOR alone, wider than the 2.0x signal the assertion has to
+      resolve. It would manufacture flake rather than absorb it.
+
+    WHAT REPLACES IT. The baseline is the SAME quantity as the numerator, taken
+    on the same loop in the same process in an adjacent window, differing only in
+    whether the operation under test was in flight. Under an external delay L the
+    baseline reads `b + L` and the measurement `b + L + tax`, so the ratio is
+    `1 + tax/(b + L)` — decreasing in L. Load pushes it toward 1.0 and away from
+    the budget, and the regression it guards raises only the numerator.
+
+    THE PREMISE THAT ALGEBRA RESTS ON, stated because it is not guaranteed: L
+    must be COMPARABLE across the baseline and measurement windows. They are
+    temporally disjoint — a caller typically spans a second or two end to end —
+    so a load burst from a sibling job that lands inside the measurement window
+    and is gone before the baseline is resampled inflates the numerator with no
+    matching rise in the denominator, and pushes the ratio UP. That is a real
+    false-failure path this helper does not exclude. It is much narrower than the
+    absolute wall-clock ceiling it replaces, which tripped on any load bump at
+    all: a spike now has to be timed to miss the baseline windows entirely. But
+    the guarantee is "much less likely", NOT "impossible", and a red here is not
+    on its own proof of a regression. Callers should bracket the measurement with
+    a baseline window either side rather than take one before it, so that a
+    monotone drift in load is shared by both terms.
+
+    TWO MORE THINGS A CALLER OWNS, not this helper:
+
+    * A SETTLING GAP between the operation and a trailing baseline window. If the
+      operation's effects outlive its completion — pool teardown, a GC pause,
+      page-cache pressure — that tax lands in the "idle" baseline, inflating the
+      denominator and SHRINKING the ratio. That weakens detection exactly where
+      it matters, and it fails quietly. #15221's caller samples with no gap
+      because its workload demonstrably leaves nothing behind; a caller whose
+      does should wait it out first.
+    * The STATISTIC over each window. A median answers "did the typical sample
+      move", so a regression that delays only a minority of samples will not move
+      it. Choose a high percentile instead where a tail is the thing at stake.
+
+    Fails on four distinct conditions, all of them real: the measurement exceeded
+    its budget relative to its own baseline; nothing was measured; no baseline
+    was measured; or the baseline is zero and cannot be a divisor.
+    """
+    assert isinstance(measured_ms, (int, float)), f"{what}: no measurement was taken (got {measured_ms!r})"
+    assert isinstance(baseline_ms, (int, float)), f"{what}: no baseline was taken (got {baseline_ms!r})"
+    assert measured_ms > 0.0, f"{what}: measured {measured_ms}ms — the operation under test did not run"
+    assert baseline_ms > 0.0, f"{what}: baseline measured {baseline_ms}ms and cannot be a divisor"
+
+    ratio = measured_ms / baseline_ms
+    report = (
+        f"{what}: {ratio:.3f}x its own idle baseline (budget {budget_ratio}) "
+        f"= {measured_ms:.3f}ms / {baseline_ms:.3f}ms idle"
+    )
+    print(f"[perf #15221] {report}")  # noqa: print
+    recorder = _WORK_UNIT_RECORDER.get()
+    if recorder is not None:
+        recorder("perf_baseline_ratio", report)
+
+    assert ratio < budget_ratio, (
+        f"{report}. This is a ratio against a baseline measured on the same loop "
+        f"moments earlier, not a wall-clock ceiling — runner load moves both "
+        f"terms, so investigate the code, do not raise it."
+    )