from claude_matrix_bot.reset_watcher.diff import ( Alert, State, WindowSnapshot, WindowState, compute_alerts, ) WEEKLY = ("seven_day", "seven_day_opus", "seven_day_sonnet") THRESHOLDS = (50, 80) def _fresh(util_by_window: dict[str, tuple[float, int]]) -> dict[str, WindowSnapshot]: return { name: WindowSnapshot(utilization=util, resets_at=reset) for name, (util, reset) in util_by_window.items() } def test_cold_start_emits_no_alerts_and_records_initial_state() -> None: fresh = _fresh( { "five_hour": (0.42, 1_000_000), "seven_day": (0.55, 2_000_000), "seven_day_opus": (0.10, 2_000_000), "seven_day_sonnet": (0.62, 2_000_000), } ) alerts, new_state = compute_alerts( old_state=None, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert alerts == [] assert new_state.windows["seven_day"].resets_at == 2_000_000 assert new_state.windows["seven_day"].utilization == 0.55 assert new_state.windows["seven_day"].alerted_thresholds == [] def test_no_changes_no_alerts() -> None: old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.40, alerted_thresholds=[]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.43, 1_000_000), "seven_day": (0.41, 2_000_000), "seven_day_opus": (0.11, 2_000_000), "seven_day_sonnet": (0.21, 2_000_000), } ) alerts, _ = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert alerts == [] def test_natural_rollover_does_not_alert() -> None: old = State( version=1, last_run_ts=999_999, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.90, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.30, alerted_thresholds=[50]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.05, 1_018_000), # new window after old expired "seven_day": (0.31, 2_000_000), "seven_day_opus": (0.11, 2_000_000), "seven_day_sonnet": (0.21, 2_000_000), } ) alerts, new_state = compute_alerts( old_state=old, fresh=fresh, now=1_010_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert alerts == [] assert new_state.windows["five_hour"].resets_at == 1_018_000 def test_mid_window_reset_shift_alerts() -> None: old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.40, alerted_thresholds=[]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.42, 1_005_000), # shifted while still in old window "seven_day": (0.40, 2_000_000), "seven_day_opus": (0.10, 2_000_000), "seven_day_sonnet": (0.20, 2_000_000), } ) alerts, _ = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert len(alerts) == 1 assert alerts[0].kind == "reset_shift" assert alerts[0].window == "five_hour" assert alerts[0].detail["old_resets_at"] == 1_000_000 assert alerts[0].detail["new_resets_at"] == 1_005_000 def test_weekly_crosses_50_percent_alerts_once() -> None: old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.49, alerted_thresholds=[]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.43, 1_000_000), "seven_day": (0.51, 2_000_000), "seven_day_opus": (0.11, 2_000_000), "seven_day_sonnet": (0.21, 2_000_000), } ) alerts, new_state = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert [a.kind for a in alerts] == ["threshold_crossed"] assert alerts[0].window == "seven_day" assert alerts[0].detail["threshold"] == 50 assert alerts[0].detail["utilization_pct"] == 51.0 assert 50 in new_state.windows["seven_day"].alerted_thresholds def test_threshold_already_alerted_does_not_refire() -> None: old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.55, alerted_thresholds=[50]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.43, 1_000_000), "seven_day": (0.60, 2_000_000), "seven_day_opus": (0.11, 2_000_000), "seven_day_sonnet": (0.21, 2_000_000), } ) alerts, _ = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert alerts == [] def test_jumps_past_both_thresholds_in_one_step_fires_both() -> None: old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.43, 1_000_000), "seven_day": (0.85, 2_000_000), "seven_day_opus": (0.11, 2_000_000), "seven_day_sonnet": (0.21, 2_000_000), } ) alerts, new_state = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) thresholds_fired = sorted(a.detail["threshold"] for a in alerts if a.window == "seven_day") assert thresholds_fired == [50, 80] assert sorted(new_state.windows["seven_day"].alerted_thresholds) == [50, 80] def test_window_rollover_resets_alerted_thresholds() -> None: old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.95, alerted_thresholds=[50, 80]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.43, 1_000_000), "seven_day": (0.05, 2_604_800), # new weekly window, fresh "seven_day_opus": (0.01, 2_604_800), "seven_day_sonnet": (0.02, 2_604_800), } ) alerts, new_state = compute_alerts( old_state=old, fresh=fresh, now=2_100_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert alerts == [] assert new_state.windows["seven_day"].alerted_thresholds == [] assert new_state.windows["seven_day"].resets_at == 2_604_800 def test_five_hour_utilization_does_not_trigger_threshold_alert() -> None: old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.95, 1_000_000), # 5h at 95% — should NOT alert "seven_day": (0.11, 2_000_000), "seven_day_opus": (0.11, 2_000_000), "seven_day_sonnet": (0.21, 2_000_000), } ) alerts, _ = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert alerts == [] def test_all_three_weekly_buckets_evaluated_independently() -> None: old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.49, alerted_thresholds=[]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.79, alerted_thresholds=[50]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.45, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.43, 1_000_000), "seven_day": (0.51, 2_000_000), # crosses 50 "seven_day_opus": (0.81, 2_000_000), # crosses 80 (50 already alerted) "seven_day_sonnet": (0.46, 2_000_000), # still under 50 } ) alerts, _ = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) fired = {(a.window, a.detail["threshold"]) for a in alerts} assert fired == {("seven_day", 50), ("seven_day_opus", 80)} def test_new_window_starting_above_threshold_alerts_for_that_threshold() -> None: """If a fresh weekly window begins (e.g. plan switch) and current util is already > 50%, we should still emit the 50% alert — otherwise we'd silently miss it.""" old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.95, alerted_thresholds=[50, 80]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.43, 1_000_000), "seven_day": (0.60, 2_604_800), # fresh window, but already at 60% "seven_day_opus": (0.11, 2_604_800), "seven_day_sonnet": (0.21, 2_604_800), } ) alerts, new_state = compute_alerts( old_state=old, fresh=fresh, now=2_100_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert any(a.window == "seven_day" and a.detail["threshold"] == 50 for a in alerts) assert 50 in new_state.windows["seven_day"].alerted_thresholds def _state_with_five_hour(reset: int) -> State: return State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=reset, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.40, alerted_thresholds=[]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) def test_sub_tolerance_reset_shift_is_suppressed() -> None: old = _state_with_five_hour(1_000_000) fresh = _fresh( { "five_hour": (0.42, 1_000_060), # +60s, default tolerance 120s "seven_day": (0.40, 2_000_000), "seven_day_opus": (0.10, 2_000_000), "seven_day_sonnet": (0.20, 2_000_000), } ) alerts, _ = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert alerts == [] def test_above_tolerance_reset_shift_alerts() -> None: old = _state_with_five_hour(1_000_000) fresh = _fresh( { "five_hour": (0.42, 1_000_300), # +300s, way above default 120s "seven_day": (0.40, 2_000_000), "seven_day_opus": (0.10, 2_000_000), "seven_day_sonnet": (0.20, 2_000_000), } ) alerts, _ = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) assert [a.kind for a in alerts] == ["reset_shift"] assert alerts[0].detail["delta_seconds"] == 300 def test_explicit_tolerance_override() -> None: """Custom tolerance lets a 60s shift fire if the operator wants stricter detection.""" old = _state_with_five_hour(1_000_000) fresh = _fresh( { "five_hour": (0.42, 1_000_060), "seven_day": (0.40, 2_000_000), "seven_day_opus": (0.10, 2_000_000), "seven_day_sonnet": (0.20, 2_000_000), } ) alerts, _ = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, reset_shift_tolerance_sec=30, ) assert [a.kind for a in alerts] == ["reset_shift"] def test_sub_tolerance_jitter_preserves_threshold_dedup() -> None: """A 60s minute-boundary jitter must NOT reset alerted_thresholds, otherwise 50%/80% would re-fire on every poll that crosses a minute boundary.""" old = State( version=1, last_run_ts=998_000, windows={ "five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]), "seven_day": WindowState(resets_at=2_000_000, utilization=0.55, alerted_thresholds=[50]), "seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]), "seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]), }, ) fresh = _fresh( { "five_hour": (0.43, 1_000_000), "seven_day": (0.60, 2_000_060), # +60s sub-tolerance jitter on weekly "seven_day_opus": (0.11, 2_000_000), "seven_day_sonnet": (0.21, 2_000_000), } ) alerts, new_state = compute_alerts( old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY, ) # No re-fire of 50% threshold. assert alerts == [] # Dedup state carried across the jitter. assert 50 in new_state.windows["seven_day"].alerted_thresholds