fix(anomalies): a level over its usual hold opens an anomaly only past its usual peak or once held (#6723)

Rows on the live instance were mostly bursts of a few minutes: 28 of 52
peaked within the metric's own usual peak, and 42 of 52 latest occurrences
lasted 30 minutes or less, which is what a burst shorter than the 15-minute
average looks like once its span is back-dated. Nothing asked how high or how
long, as the System 182 ruling says a normal records.

A departure now also has to be past the slot's usual peak, or held: the
median of the last hour over the same line. Those spans were also keeping the
burstiest hours out of learning, lowering the usual lines further.

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

steward/core/anomalies.py

@@ -2,8 +2,12 @@
 
 Every minute the last WINDOW of each learned key is averaged and compared
 with the normal for the slot it falls in (steward.core.norms). A level above
-the slot's usual hold by DEPART times, and by at least the metric's floor,
-opens an occurrence; the occurrence ends when the level comes back.
+the slot's usual hold by DEPART times, and by at least the metric's floor, has
+departed. That opens an occurrence once it is also outside what the normal
+records in height or in time: higher than the slot's usual peak, or held, with
+most of the last HELD above the line. A burst of a few minutes within the
+usual peak lifts the average for a while and opens nothing. 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"
@@ -28,6 +32,7 @@ from steward.timeline.record import anomaly_changed
 
 WINDOW = timedelta(minutes=15)   # the level judged is this window's average
 DEPART = 1.25                    # times the slot's usual hold that counts as a departure
+HELD = timedelta(hours=1)        # a level within the usual peak opens once the median of this is out
 # An occurrence that comes back within one window of ending continues the
 # same occurrence: a level wobbling around the line is one event, not ten.
 REJOIN = WINDOW
@@ -118,6 +123,32 @@ 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
+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
+    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
+""").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."""
+    if not keys:
+        return {}
+    rows = await session.execute(_HELD_LEVELS, {
+        "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}
+
+
 async def detect(session, now: datetime | None = None) -> dict:
     """One pass: open, extend and close occurrences. Runs inside the caller's
     transaction. Returns {"opened", "closed"} counts."""
@@ -135,10 +166,13 @@ 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_peak], now)
     for key, level in levels.items():
         norm = norms.get(key)
-        learned = LEARNED.get((key[0], key[2]))
-        out = norm is not None and learned is not None and departs(level, norm, learned.least_departure)
+        out = key in out_now
         current = open_now.pop(key, None)
         if current is not None:
             span, anomaly = current
@@ -148,8 +182,11 @@ async def detect(session, now: datetime | None = None) -> dict:
                 _close(anomaly, span, now)
                 changed.append(anomaly)
                 closed += 1
-        elif out:
-            changed.append(await _occur(session, key, level, norm, now))
+        elif out and level > norm.usual_peak:
+            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):
+            changed.append(await _occur(session, key, level, norm, now, HELD))
             opened += 1
     # An open occurrence whose key stopped reporting, lost its normal or fell
     # below its gate can't be judged any more: it ends now.
@@ -175,11 +212,12 @@ def _close(anomaly: MetricAnomaly, span: MetricAnomalySpan, now: datetime) -> No
     anomaly.ended_at = now
 
 
-async def _occur(session, key, level: float, norm: Norm, now: datetime) -> MetricAnomaly:
+async def _occur(session, key, level: float, norm: Norm, now: datetime, judged_over: timedelta) -> MetricAnomaly:
     """A departure on a key with no open occurrence: join the key's anomaly
-    still to review, or open a new one. Returns the anomaly."""
+    still to review, or open a new one, starting where the stretch it was
+    judged over starts. Returns the anomaly."""
     source, resource, metric = key
-    start = now - WINDOW
+    start = now - judged_over
     anomaly = (await session.execute(
         select(MetricAnomaly)
         .where(MetricAnomaly.source_module == source, MetricAnomaly.resource_name == resource,