diff --git a/.agents/skills/shared/metric-alarms.toml b/.agents/skills/shared/metric-alarms.toml index 764b03a6..9f3aeed2 100644 --- a/.agents/skills/shared/metric-alarms.toml +++ b/.agents/skills/shared/metric-alarms.toml @@ -748,6 +748,14 @@ labels = [{ key = "reason", value = "forcing_catchup_behind" }] delta_threshold = 1 streak_threshold = 3 burst_threshold = 10 +# Design tick cadence the streak_threshold was calibrated against (#3790). The +# streak dwell is time-denominated: recovery-stalled fires only once the streak +# count is met AND now - first_breach_ts >= streak_threshold * +# expected_interval_seconds (3 * 1200 = 3600s). Set to the DESIGN cadence +# (below the real ~44 min tick cadence) so the healthy sustained case still +# fires on the same tick as before; only a compressed/duplicate cadence that +# would advance the streak at 2x wall-clock speed is held. +expected_interval_seconds = 1200 # Post-restart absolute-value fire (#3198). On a PID/start_ticks change (restart) # or a counter reset the streak baseline is re-collected and the cross-tick delta # machine is blind to a stall that accrued during startup/warmup. If the first @@ -756,8 +764,8 @@ burst_threshold = 10 # episode accrued 213; a clean restart reaching real-time sync promptly accrues # only a handful. 0/absent disables the check. post_restart_absolute_threshold = 50 -baseline_version = 3 # bumped in #3222: de-double snapshot_file (metrics/ prefix removed → bare filename) -semantic_change_date = "2026-06-07T00:00:00Z" +baseline_version = 4 # bumped in #3790: streak dwell now time-denominated (expected_interval_seconds added; fires on wall-clock dwell, not tick count) — invalidates prior acks +semantic_change_date = "2026-08-23T00:00:00Z" # time-denominated streak dwell per #3790 severity = "WARN" gates = ["validator-only"] snapshot_file = "counter_streak_snapshot" diff --git a/.claude/skills/shared/metric-alarms.toml b/.claude/skills/shared/metric-alarms.toml index 764b03a6..9f3aeed2 100644 --- a/.claude/skills/shared/metric-alarms.toml +++ b/.claude/skills/shared/metric-alarms.toml @@ -748,6 +748,14 @@ labels = [{ key = "reason", value = "forcing_catchup_behind" }] delta_threshold = 1 streak_threshold = 3 burst_threshold = 10 +# Design tick cadence the streak_threshold was calibrated against (#3790). The +# streak dwell is time-denominated: recovery-stalled fires only once the streak +# count is met AND now - first_breach_ts >= streak_threshold * +# expected_interval_seconds (3 * 1200 = 3600s). Set to the DESIGN cadence +# (below the real ~44 min tick cadence) so the healthy sustained case still +# fires on the same tick as before; only a compressed/duplicate cadence that +# would advance the streak at 2x wall-clock speed is held. +expected_interval_seconds = 1200 # Post-restart absolute-value fire (#3198). On a PID/start_ticks change (restart) # or a counter reset the streak baseline is re-collected and the cross-tick delta # machine is blind to a stall that accrued during startup/warmup. If the first @@ -756,8 +764,8 @@ burst_threshold = 10 # episode accrued 213; a clean restart reaching real-time sync promptly accrues # only a handful. 0/absent disables the check. post_restart_absolute_threshold = 50 -baseline_version = 3 # bumped in #3222: de-double snapshot_file (metrics/ prefix removed → bare filename) -semantic_change_date = "2026-06-07T00:00:00Z" +baseline_version = 4 # bumped in #3790: streak dwell now time-denominated (expected_interval_seconds added; fires on wall-clock dwell, not tick count) — invalidates prior acks +semantic_change_date = "2026-08-23T00:00:00Z" # time-denominated streak dwell per #3790 severity = "WARN" gates = ["validator-only"] snapshot_file = "counter_streak_snapshot" diff --git a/scripts/ci/check-alarm-versions.py b/scripts/ci/check-alarm-versions.py index 52349f72..c1b91a38 100644 --- a/scripts/ci/check-alarm-versions.py +++ b/scripts/ci/check-alarm-versions.py @@ -29,6 +29,7 @@ "op", "threshold", "multiplier", "min_absolute", "gates", "for_ticks", "burst_threshold", "delta_threshold", "streak_threshold", + "expected_interval_seconds", "post_restart_absolute_threshold", "denominator", "denominator_metric", "denominator_sum", "denominator_extraction", "denominator_includes_numerator", diff --git a/scripts/lib/eval-alarms.py b/scripts/lib/eval-alarms.py index ea61b52b..f77ea010 100644 --- a/scripts/lib/eval-alarms.py +++ b/scripts/lib/eval-alarms.py @@ -1482,11 +1482,24 @@ def eval_counter_streak( start_ticks: str, gap_stale: bool = False, too_fresh: bool = False, + now: int | None = None, ) -> dict: """Evaluate a counter-streak alarm. Independent of PREV_PROM_INVALID — uses own PID/start_ticks in snapshot. + + The streak dwell is TIME-denominated, not tick-denominated (#3790): the + streak-opening tick records `first_breach_ts`, and the alarm fires only once + `streak >= streak_threshold` AND `now - first_breach_ts` has reached + `streak_threshold * expected_interval_seconds`. This makes the confirmation + gate independent of tick cadence — a natural interval split into two + sub-window halves (each above the #3757 too-fresh floor, so that guard + misses them) can no longer advance the streak to its firing threshold at 2x + wall-clock speed. `now` defaults to int(time.time()); it is a parameter for + deterministic tests. """ + if now is None: + now = int(time.time()) ev_default = default_extra_values(alarm, "counter-streak") # Missing process identity guard (#3279): an abbreviated tick that skips @@ -1622,10 +1635,40 @@ def eval_counter_streak( delta_threshold = alarm.get("delta_threshold", 1) streak_threshold = alarm.get("streak_threshold", 3) burst_threshold = alarm.get("burst_threshold", 10) + # The DESIGN cadence the streak_threshold was calibrated against (#3790): + # streak_threshold * expected_interval_seconds is the wall-clock dwell a + # sustained breach must survive before firing. Deliberately the *design* + # cadence (below the real ~44 min tick cadence), so the healthy sustained + # case still fires on the same tick as before; only a compressed/duplicate + # cadence that would advance the count at 2x wall-clock speed is held. + expected_interval_seconds = alarm.get("expected_interval_seconds", 1200) + + # first_breach_ts anchors the streak to wall-clock time. It is set on the + # streak-opening (0->1) tick and preserved verbatim as the streak advances. + # A legacy snapshot carrying breach_streak>=1 but no first_breach_ts (written + # before #3790) re-anchors to `now` here — a one-time conservative delay, + # never a spurious fire off a stale unanchored streak. Every reset branch + # above rewrites the snapshot WITHOUT this key, which clears it. + prev_first_breach_ts_str = snapshot.get("first_breach_ts") + + def _anchor(prev_streak: int) -> int: + """Resolve first_breach_ts for a breaching tick. + + Anchor to `now` when the streak is opening (prev_streak == 0) or when a + legacy snapshot advanced a streak with no persisted anchor; otherwise + preserve the existing anchor. + """ + if prev_streak == 0 or prev_first_breach_ts_str is None: + return now + try: + return int(prev_first_breach_ts_str) + except ValueError: + return now ev = {"streak": streak, "streak_threshold": streak_threshold} if delta >= burst_threshold: + first_breach_ts = _anchor(streak) streak += 1 new_snapshot = { "version": "1", @@ -1633,14 +1676,17 @@ def eval_counter_streak( "start_ticks": start_ticks, "counter_value": str(int(cur_val)), "breach_streak": str(streak), + "first_breach_ts": str(first_breach_ts), } write_snapshot(snapshot_path, new_snapshot) + # Acute burst is a single-tick spike — fire immediately, ungated by dwell. return make_result( alarm, "firing", value=delta, threshold=burst_threshold, for_ticks_elapsed=streak, extra_values={"streak": streak, "streak_threshold": streak_threshold}, ) if delta >= delta_threshold: + first_breach_ts = _anchor(streak) streak += 1 new_snapshot = { "version": "1", @@ -1648,20 +1694,34 @@ def eval_counter_streak( "start_ticks": start_ticks, "counter_value": str(int(cur_val)), "breach_streak": str(streak), + "first_breach_ts": str(first_breach_ts), } write_snapshot(snapshot_path, new_snapshot) - if streak >= streak_threshold: + dwell_required = streak_threshold * expected_interval_seconds + dwell_elapsed = now - first_breach_ts + ev_streak = { + "streak": streak, + "streak_threshold": streak_threshold, + "dwell_elapsed": dwell_elapsed, + "dwell_required": dwell_required, + } + # Fire only when BOTH the count AND the wall-clock dwell are satisfied. + # If the count is met but the dwell is not, keep dwelling (breach): the + # snapshot (with first_breach_ts) is preserved because a "breach" return + # never triggers maybe_reset_counter_snapshot (which only acts on + # "skipped"). + if streak >= streak_threshold and dwell_elapsed >= dwell_required: return make_result( alarm, "firing", value=delta, threshold=delta_threshold, - for_ticks_elapsed=streak, extra_values={"streak": streak, "streak_threshold": streak_threshold}, + for_ticks_elapsed=streak, extra_values=ev_streak, ) return make_result( alarm, "breach", value=delta, threshold=delta_threshold, - for_ticks_elapsed=streak, extra_values={"streak": streak, "streak_threshold": streak_threshold}, + for_ticks_elapsed=streak, extra_values=ev_streak, ) - # delta == 0 + # delta == 0 — reset the streak and clear the wall-clock anchor. new_snapshot = { "version": "1", "pid": pid, @@ -2049,6 +2109,12 @@ def main() -> int: alarms = catalog.get("alarm", []) results: list[dict] = [] + # Single wall-clock reference for the whole tick: the counter-streak + # time-dwell gate (#3790) and the cooldown-window math below share it, so + # both see the same `now`. `--now` overrides int(time.time()) for + # deterministic tests. Nothing between here and either consumer mutates it. + now = args.now if args.now is not None else int(time.time()) + for alarm in alarms: name = alarm["name"] kind = alarm["kind"] @@ -2089,7 +2155,7 @@ def main() -> int: ) elif kind == "counter-streak": result = eval_counter_streak(alarm, current, state_dir, pid, start_ticks_val, - gap_stale, too_fresh) + gap_stale, too_fresh, now=now) else: result = make_result(alarm, "skipped", skip_reason=f"unknown kind: {kind}") @@ -2119,7 +2185,6 @@ def main() -> int: # owns read/suppress/write so the manual JSON edit can be deleted. Absent # --cooldown-file, this is a no-op and behavior is byte-for-byte legacy. if args.cooldown_file: - now = args.now if args.now is not None else int(time.time()) cooldown_path = Path(args.cooldown_file) cooldown_data = read_cooldown_file(cooldown_path) apply_cooldowns(results, cooldown_data, now) diff --git a/scripts/lib/test_eval_alarms_streak_dwell.py b/scripts/lib/test_eval_alarms_streak_dwell.py new file mode 100644 index 00000000..fbe22495 --- /dev/null +++ b/scripts/lib/test_eval_alarms_streak_dwell.py @@ -0,0 +1,298 @@ +#!/usr/bin/env python3 +"""Regression + coverage tests for the time-denominated counter-streak dwell (#3790). + +Follow-up from #3757. The too-fresh guard (#3757) fixes the destructive *reset* +direction (a sub-MIN_EVAL_WINDOW duplicate that samples delta=0 and zeros the +streak). It does NOT fix the *acceleration* direction ("vacuous dwell"): when a +natural ~38 min interval is split into two ~19 min halves that each independently +clear delta>=1, both exceed the 600s too-fresh floor, so breach_streak still +advances at 2x wall-clock speed and `streak >= streak_threshold` fires early. + +Fix: make the streak dwell TIME-denominated rather than tick-denominated. Store +`first_breach_ts` on the streak-opening tick and require +`now - first_breach_ts >= streak_threshold * expected_interval_seconds` IN +ADDITION to the count before firing. This makes the confirmation gate +independent of tick cadence and survives a future cadence change. +""" + +import sys +import tempfile +from pathlib import Path + +# eval-alarms.py uses a hyphen, so we need importlib +import importlib.util + +_spec = importlib.util.spec_from_file_location( + "eval_alarms", + Path(__file__).parent / "eval-alarms.py", +) +_mod = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_mod) + +read_snapshot = _mod.read_snapshot +write_snapshot = _mod.write_snapshot +maybe_reset_counter_snapshot = _mod.maybe_reset_counter_snapshot +eval_counter_streak = _mod.eval_counter_streak + + +def _make_alarm(name="recovery-stalled", **kwargs): + """Create a minimal counter-streak alarm dict for testing.""" + alarm = { + "name": name, + "kind": "counter-streak", + "metric": "m", + "delta_threshold": 1, + "streak_threshold": 3, + "burst_threshold": 10, + "severity": "WARN", + "expected_interval_seconds": 1200, + } + alarm.update(kwargs) + return alarm + + +def _seed(state_dir, *, counter_value, breach_streak, first_breach_ts=None, + pid="P", start_ticks="T"): + """Write a counter_streak_snapshot with the given fields.""" + snap = { + "version": "1", + "pid": pid, + "start_ticks": start_ticks, + "counter_value": str(counter_value), + "breach_streak": str(breach_streak), + } + if first_breach_ts is not None: + snap["first_breach_ts"] = str(first_breach_ts) + write_snapshot(state_dir / "counter_streak_snapshot", snap) + + +# ── the headline regression: 2x acceleration must not fire early ────────────── + +def test_split_interval_does_not_fire_early(): + """A ~38 min interval split into two ~19 min halves must NOT fire at streak 3. + + Red on main: no time gate — streak reaches 3 and returns "firing" at + T0+2280 (~38 min), well before the intended dwell. Green after: the count is + met but 2280 < 3*1200 = 3600, so the tick returns "breach" and keeps dwelling. + """ + T0 = 1_000_000 + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + + # Streak already opened at T0 (breach_streak=1, anchored at T0). + _seed(state_dir, counter_value=100, breach_streak=1, first_breach_ts=T0) + + # First half boundary: delta>=1 at T0+1140 (~19 min) → streak 2. + r1 = eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "P", "T", now=T0 + 1140) + assert r1["state"] == "breach", f"streak 2 must be breach, got {r1['state']}" + + # Second half boundary: delta>=1 at T0+2280 (~38 min) → streak 3, but + # 2280 < 3600 dwell → still breach, NOT firing. + r2 = eval_counter_streak(alarm, {"m": [({}, 102.0)]}, state_dir, + "P", "T", now=T0 + 2280) + assert r2["state"] == "breach", \ + f"count met but dwell unmet must be breach, got {r2['state']}" + snap = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap["breach_streak"] == "3", \ + f"streak still counts to 3, got {snap.get('breach_streak')!r}" + assert snap["first_breach_ts"] == str(T0), \ + f"first_breach_ts must be preserved, got {snap.get('first_breach_ts')!r}" + + +def test_dwell_met_fires(): + """The gate is a DELAY, not a permanent suppression: once wall-clock dwell is + satisfied, streak >= threshold fires.""" + T0 = 1_000_000 + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + _seed(state_dir, counter_value=100, breach_streak=2, first_breach_ts=T0) + + # streak → 3 at T0+3600: 3600 >= 3*1200 → fires. + r = eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "P", "T", now=T0 + 3600) + assert r["state"] == "firing", \ + f"dwell met at streak 3 must fire, got {r['state']}" + + +# ── first_breach_ts lifecycle ───────────────────────────────────────────────── + +def test_first_breach_ts_set_on_open_and_preserved(): + """first_breach_ts is set to `now` on the 0->1 opening tick and byte-preserved + while the streak advances 1->2.""" + T0 = 2_000_000 + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + # No prior streak, no anchor. + _seed(state_dir, counter_value=100, breach_streak=0) + + # 0 -> 1: opens the streak, anchors at T0. + r1 = eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "P", "T", now=T0) + assert r1["state"] == "breach", f"streak 1 must be breach, got {r1['state']}" + snap1 = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap1["breach_streak"] == "1" + assert snap1["first_breach_ts"] == str(T0), \ + f"first_breach_ts must be set on open, got {snap1.get('first_breach_ts')!r}" + + # 1 -> 2: anchor preserved verbatim (NOT re-set to the later now). + r2 = eval_counter_streak(alarm, {"m": [({}, 102.0)]}, state_dir, + "P", "T", now=T0 + 5000) + assert r2["state"] == "breach" + snap2 = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap2["breach_streak"] == "2" + assert snap2["first_breach_ts"] == str(T0), \ + f"first_breach_ts must be preserved on advance, got {snap2.get('first_breach_ts')!r}" + + +def test_first_breach_ts_cleared_on_reset(): + """Every reset branch (delta==0, counter-reset, PID-change, gap-stale) drops + both breach_streak and first_breach_ts.""" + T0 = 3_000_000 + + # delta == 0 + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + _seed(state_dir, counter_value=100, breach_streak=2, first_breach_ts=T0) + eval_counter_streak(alarm, {"m": [({}, 100.0)]}, state_dir, + "P", "T", now=T0 + 1200) + snap = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap["breach_streak"] == "0", "delta==0 must zero the streak" + assert "first_breach_ts" not in snap, \ + f"delta==0 must clear first_breach_ts, got {snap.get('first_breach_ts')!r}" + + # counter reset (cur < prev) + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + _seed(state_dir, counter_value=100, breach_streak=2, first_breach_ts=T0) + eval_counter_streak(alarm, {"m": [({}, 50.0)]}, state_dir, + "P", "T", now=T0 + 1200) + snap = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap["breach_streak"] == "0" + assert "first_breach_ts" not in snap, \ + "counter-reset must clear first_breach_ts" + + # PID change + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + _seed(state_dir, counter_value=100, breach_streak=2, first_breach_ts=T0, + pid="OLD") + eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "NEW", "T", now=T0 + 1200) + snap = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap["breach_streak"] == "0" + assert "first_breach_ts" not in snap, \ + "PID-change re-baseline must clear first_breach_ts" + + # gap-stale + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + _seed(state_dir, counter_value=100, breach_streak=2, first_breach_ts=T0) + eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "P", "T", gap_stale=True, now=T0 + 1200) + snap = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap["breach_streak"] == "0" + assert "first_breach_ts" not in snap, \ + "gap-stale re-baseline must clear first_breach_ts" + + +def test_legacy_snapshot_without_first_breach_ts_reanchors(): + """A legacy snapshot with breach_streak>=1 but NO first_breach_ts must + re-anchor to `now` on the next tick and NOT fire off the stale unanchored + streak (one-time conservative delay, never a spurious fire).""" + T0 = 4_000_000 + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + # Legacy: streak already at 2, but no anchor persisted. + _seed(state_dir, counter_value=100, breach_streak=2) + + r = eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "P", "T", now=T0) + assert r["state"] == "breach", \ + f"unanchored streak must not fire on re-anchor tick, got {r['state']}" + snap = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap["breach_streak"] == "3" + assert snap["first_breach_ts"] == str(T0), \ + f"legacy streak must re-anchor first_breach_ts to now, got {snap.get('first_breach_ts')!r}" + + +def test_dwell_preserved_across_reset_hook(): + """A sub-dwell breach tick returns state="breach"; the centralized + maybe_reset_counter_snapshot hook only mutates on state=="skipped", so a + "breach" return preserves first_breach_ts end-to-end (Critic A invariant).""" + T0 = 5_000_000 + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + _seed(state_dir, counter_value=100, breach_streak=1, first_breach_ts=T0) + + r = eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "P", "T", now=T0 + 1140) + assert r["state"] == "breach" + + # Drive the same post-processing hook main() calls after each result. + maybe_reset_counter_snapshot(alarm, "counter-streak", r["state"], + state_dir, r.get("skip_reason")) + snap = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap["first_breach_ts"] == str(T0), \ + f"reset hook must preserve first_breach_ts on breach, got {snap.get('first_breach_ts')!r}" + assert snap["breach_streak"] == "2", \ + f"reset hook must preserve breach_streak on breach, got {snap.get('breach_streak')!r}" + + +# ── burst path stays ungated ────────────────────────────────────────────────── + +def test_burst_still_fires_immediately(): + """The acute burst path (delta >= burst_threshold) fires on a single tick + regardless of dwell, but still initializes first_breach_ts.""" + T0 = 6_000_000 + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + _seed(state_dir, counter_value=100, breach_streak=0) + + # delta=15 >= burst_threshold=10 on the opening tick. + r = eval_counter_streak(alarm, {"m": [({}, 115.0)]}, state_dir, + "P", "T", now=T0) + assert r["state"] == "firing", \ + f"burst must fire immediately, got {r['state']}" + snap = read_snapshot(state_dir / "counter_streak_snapshot") + assert snap["first_breach_ts"] == str(T0), \ + f"burst must anchor first_breach_ts, got {snap.get('first_breach_ts')!r}" + + +def test_expected_interval_default(): + """With no expected_interval_seconds in the alarm, the 1200s default is used, + so the dwell required for streak 3 is 3*1200 = 3600s.""" + T0 = 7_000_000 + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) + alarm = _make_alarm() + del alarm["expected_interval_seconds"] # exercise the default + + # streak 2 -> 3 just below 3600s dwell → breach. + _seed(state_dir, counter_value=100, breach_streak=2, first_breach_ts=T0) + r_below = eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "P", "T", now=T0 + 3599) + assert r_below["state"] == "breach", \ + f"3599 < 3600 default dwell must be breach, got {r_below['state']}" + + # And exactly at 3600s → firing. + _seed(state_dir, counter_value=100, breach_streak=2, first_breach_ts=T0) + r_at = eval_counter_streak(alarm, {"m": [({}, 101.0)]}, state_dir, + "P", "T", now=T0 + 3600) + assert r_at["state"] == "firing", \ + f"3600 == default dwell must fire, got {r_at['state']}" + + +if __name__ == "__main__": + import pytest + sys.exit(pytest.main([__file__, "-v"]))