fix(anomalies): one slow reading no longer opens an anomaly (#6793)

A single tall reading lifted a 15-minute average past the usual height
on its own, so one slow request opened a row that stayed open until the
reading rolled out of the window. Four of the seven rows that occurred
after the 10-30 detection change were this: HA web (one 6.7 s check),
error-pages, codex and mailcow-svc latency.

detect() now opens by height only when the window is still past the
usual height, and still departs, with its highest reading left out. A
window that fails that falls through to the held test, which now needs
at least HELD_FEWEST (3) readings behind its median, so a lone reading
on a quiet router opens nothing either way.

The median and the topless average share one per-key read (_per_key).
The architecture doc and get_metric_normal's description state the
rule.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/core/anomalies.py

@@ -7,7 +7,10 @@ departed. That opens an occurrence once it is also outside what the normal
 records in height or in time: higher than the metric usually goes (Norm.
 usual_height), or held, with most of the last HELD above the line. A burst of
 a few minutes within that lifts the average for a while and opens nothing. The
-occurrence ends when the level comes back under the line.
+height is judged with the window's highest reading left out, so one slow
+request or one busy sample can't carry a window past it, and a held level needs
+at least HELD_FEWEST readings behind its median: a lone reading never opens
+one. The occurrence ends when the level comes back under the line.
 
 What the operator ruled (System 182 "Metric Norms and Anomalies"):
 - A departure is a warning, marked as a metric anomaly, and stays "to review"
@@ -123,30 +126,58 @@ async def levels_now(session, now: datetime) -> dict[tuple[str, str, str], float
     return {(s, r, m): float(level) for s, r, m, level in rows}
 
 
-# The median, as a burst moves an average of the same stretch but not this.
-_HELD_LEVELS = text("""
-SELECT k.s, k.r, k.m, v.held
+def _per_key(aggregate: str):
+    """A read of `aggregate` over each given key's readings in [:since, :now].
+    A key whose aggregate is NULL is left out."""
+    return text(f"""
+SELECT k.s, k.r, k.m, v.x
 FROM unnest(:sources, :resources, :metrics) AS k(s, r, m)
 CROSS JOIN LATERAL (
-    SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY p.value) AS held FROM plugin_metrics p
+    SELECT {aggregate} AS x FROM plugin_metrics p
     WHERE p.source_module = k.s AND p.resource_name = k.r AND p.metric_name = k.m
       AND p.recorded_at >= :since AND p.recorded_at <= :now
 ) v
-WHERE v.held IS NOT NULL
+WHERE v.x IS NOT NULL
 """).bindparams(bindparam("sources", type_=ARRAY(String)), bindparam("resources", type_=ARRAY(String)),
                 bindparam("metrics", type_=ARRAY(String)))
 
 
-async def held_levels(session, keys: list[tuple[str, str, str]], now: datetime) -> dict[tuple[str, str, str], float]:
-    """{(source, resource, metric): the median of its last HELD}. Asked only of
-    keys that have departed and stayed within their usual peak, so the longer
-    read isn't paid for the rest."""
+# The median, as a burst moves an average of the same stretch but not this.
+# Fewer than HELD_FEWEST readings in the stretch are not a held level.
+HELD_FEWEST = 3
+_HELD_LEVELS = _per_key(
+    f"CASE WHEN count(*) >= {HELD_FEWEST} THEN percentile_cont(0.5) WITHIN GROUP (ORDER BY p.value) END")
+# The average with the single highest reading left out; NULL for one reading.
+_TOPLESS_LEVELS = _per_key("CASE WHEN count(*) > 1 THEN (sum(p.value) - max(p.value)) / (count(*) - 1) END")
+
+
+async def _read_per_key(session, query, keys: list[tuple[str, str, str]], since: datetime,
+                        now: datetime) -> dict[tuple[str, str, str], float]:
     if not keys:
         return {}
-    rows = await session.execute(_HELD_LEVELS, {
+    rows = await session.execute(query, {
         "sources": [k[0] for k in keys], "resources": [k[1] for k in keys], "metrics": [k[2] for k in keys],
-        "since": now - HELD, "now": now})
-    return {(s, r, m): float(held) for s, r, m, held in rows}
+        "since": since, "now": now})
+    return {(s, r, m): float(x) for s, r, m, x in rows}
+
+
+async def held_levels(session, keys: list[tuple[str, str, str]], now: datetime) -> dict[tuple[str, str, str], float]:
+    """{(source, resource, metric): the median of its last HELD}, for keys with
+    at least HELD_FEWEST readings in it. Asked only of keys that have departed
+    and not opened by height, so the longer read isn't paid for the rest."""
+    return await _read_per_key(session, _HELD_LEVELS, keys, now - HELD, now)
+
+
+async def topless_levels(session, keys: list[tuple[str, str, str]],
+                         now: datetime) -> dict[tuple[str, str, str], float]:
+    """{(source, resource, metric): its average over the last WINDOW without
+    the window's highest reading}. A key with one reading in the window is
+    left out: one reading is never a level."""
+    return await _read_per_key(session, _TOPLESS_LEVELS, keys, now - WINDOW, now)
+
+
+def _past_height(level: float, norm: Norm, floor: float) -> bool:
+    return level > norm.usual_height and departs(level, norm, floor)
 
 
 async def detect(session, now: datetime | None = None) -> dict:
@@ -166,10 +197,17 @@ async def detect(session, now: datetime | None = None) -> dict:
     }
     opened = closed = 0
     changed: list[MetricAnomaly] = []       # opened, joined or ended: their timeline entries follow
-    out_now = {key for key, level in levels.items()
-               if key in norms and departs(level, norms[key], LEARNED[(key[0], key[2])].least_departure)}
-    held = await held_levels(session, [key for key in out_now
-                                       if key not in open_now and levels[key] <= norms[key].usual_height], now)
+    def floor(key):
+        return LEARNED[(key[0], key[2])].least_departure
+
+    out_now = {key for key, level in levels.items() if key in norms and departs(level, norms[key], floor(key))}
+    waiting = [key for key in out_now if key not in open_now]
+    # Past the usual height only if it stays past it without its highest
+    # reading: one slow request lifts a window's average on its own.
+    tall = [key for key in waiting if levels[key] > norms[key].usual_height]
+    topless = await topless_levels(session, tall, now)
+    past_height = {key for key in tall if key in topless and _past_height(topless[key], norms[key], floor(key))}
+    held = await held_levels(session, [key for key in waiting if key not in past_height], now)
     for key, level in levels.items():
         norm = norms.get(key)
         out = key in out_now
@@ -182,10 +220,10 @@ async def detect(session, now: datetime | None = None) -> dict:
                 _close(anomaly, span, now)
                 changed.append(anomaly)
                 closed += 1
-        elif out and level > norm.usual_height:
+        elif key in past_height:
             changed.append(await _occur(session, key, level, norm, now, WINDOW))
             opened += 1
-        elif out and departs(held.get(key, 0.0), norm, LEARNED[(key[0], key[2])].least_departure):
+        elif out and departs(held.get(key, 0.0), norm, floor(key)):
             changed.append(await _occur(session, key, level, norm, now, HELD))
             opened += 1
     # An open occurrence whose key stopped reporting, lost its normal or fell