fix(alerts): acknowledgements hold until resolve; clear after N OK; rule changes reset (#5059)

#5059 "Alert state machine: acknowledgement holds until resolve; N
consecutive OK to resolve; editing, disabling or deleting a rule resets it".

- The state machine is now next_step(), a pure function of one sample. An
  acknowledged alert stays acknowledged while breached; before, the next
  breaching sample re-fired it. Firing and acknowledged alerts resolve after
  consecutive_ok_required OK checks in a row (new rule column, default 1;
  rule form, rules list and MCP create/update_alert_rule clear_after_checks).
- Changing what a rule watches (source, resource, metric, operator,
  threshold) or disabling it resets its alert to inactive; a firing or
  acknowledged one records a resolved event whose new note says why, and
  nobody is notified. Web and MCP edits share alerts.service.apply_rule_change.
- Migration 0028: consecutive_ok_required, consecutive_ok_count,
  alert_events.note, and metric_value nullable (a reset has no sample).

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

tests/core/test_alert_state_machine.py

@@ -0,0 +1,57 @@
+"""The alert state machine, one sample at a time (core/alerts.next_step)."""
+import pytest
+
+from steward.core.alerts import Step, next_step
+from steward.models.alerts import AlertStateEnum as S
+
+# (state, fail, ok, breached, fail_required, ok_required) -> Step
+CASES = [
+    # inactive
+    ((S.inactive, 0, 0, False, 1, 1), Step(S.inactive, 0, 0)),
+    ((S.inactive, 0, 0, True, 1, 1), Step(S.firing, 1, 0, "firing")),
+    ((S.inactive, 0, 0, True, 3, 1), Step(S.pending, 1, 0)),
+    # pending
+    ((S.pending, 1, 0, True, 3, 1), Step(S.pending, 2, 0)),
+    ((S.pending, 2, 0, True, 3, 1), Step(S.firing, 3, 0, "firing")),
+    ((S.pending, 2, 0, False, 3, 1), Step(S.inactive, 0, 0)),
+    # firing
+    ((S.firing, 1, 0, True, 1, 1), Step(S.firing, 2, 0)),
+    ((S.firing, 1, 0, False, 1, 1), Step(S.inactive, 0, 0, "resolved")),
+    ((S.firing, 1, 0, False, 1, 3), Step(S.firing, 1, 1)),
+    ((S.firing, 1, 2, False, 1, 3), Step(S.inactive, 0, 0, "resolved")),
+    ((S.firing, 1, 2, True, 1, 3), Step(S.firing, 2, 0)),          # a breach resets the OK run
+    # acknowledged: holds while breached, however long
+    ((S.acknowledged, 1, 0, True, 1, 1), Step(S.acknowledged, 2, 0)),
+    ((S.acknowledged, 50, 0, True, 2, 1), Step(S.acknowledged, 51, 0)),
+    ((S.acknowledged, 5, 0, False, 1, 1), Step(S.inactive, 0, 0, "resolved")),
+    ((S.acknowledged, 5, 0, False, 1, 2), Step(S.acknowledged, 5, 1)),
+    ((S.acknowledged, 5, 1, True, 1, 2), Step(S.acknowledged, 6, 0)),
+    ((S.acknowledged, 5, 1, False, 1, 2), Step(S.inactive, 0, 0, "resolved")),
+    # the transient resolved state reads as inactive
+    ((S.resolved, 0, 0, True, 2, 1), Step(S.pending, 1, 0)),
+    ((S.resolved, 0, 0, False, 1, 1), Step(S.inactive, 0, 0)),
+]
+
+
+@pytest.mark.parametrize("args,want", CASES)
+def test_next_step(args, want):
+    assert next_step(*args) == want
+
+
+def test_every_state_and_outcome_is_covered():
+    seen = {(a[0], a[3]) for a, _ in CASES}
+    assert seen == {(s, b) for s in S for b in (True, False)}
+    assert {w.event for _, w in CASES} == {None, "firing", "resolved"}
+
+
+def test_an_acknowledged_alert_never_fires_again_while_breached():
+    state, fail, ok = S.acknowledged, 1, 0
+    for _ in range(100):
+        step = next_step(state, fail, ok, True, 1, 1)
+        assert step.state == S.acknowledged and step.event is None
+        state, fail, ok = step.state, step.fail_count, step.ok_count
+
+
+def test_requirements_below_one_count_as_one():
+    assert next_step(S.inactive, 0, 0, True, 0, 0) == Step(S.firing, 1, 0, "firing")
+    assert next_step(S.firing, 1, 0, False, 0, 0) == Step(S.inactive, 0, 0, "resolved")

tests/integration/test_alert_states.py

@@ -0,0 +1,107 @@
+"""Integration: alert state against real Postgres. An acknowledgement holds
+while the breach lasts, an alert clears after its OK run, and changing what a
+rule watches (or disabling it) resets its state with a note on the event."""
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def app():
+    if not os.environ.get("STEWARD_DATABASE_URL"):
+        pytest.skip("needs Postgres")
+    from steward.app import create_app
+    return create_app(testing=False)
+
+
+def test_acknowledge_holds_clear_after_and_reset(app):
+    from sqlalchemy import select
+    from steward.alerts import service as alert_svc
+    from steward.core.alerts import record_metric
+    from steward.models.alerts import AlertEvent, AlertRule, AlertState, AlertStateEnum
+    from steward.models.metrics import PluginMetric
+    from steward.models.users import User, UserRole
+
+    async def _go():
+        tag = uuid.uuid4().hex[:8]
+        res = f"box-{tag}"
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                user = User(username=f"al-{tag}", email=f"al-{tag}@test", password_hash="x",
+                            role=UserRole.admin)
+                s.add(user)
+                s.add(PluginMetric(source_module="host_agent", resource_name=res,
+                                   metric_name="cpu_pct", value=1.0))
+                await s.flush()
+                fields = await alert_svc.rule_fields(
+                    s, name=f"hot {tag}", source_module="host_agent", resource_name=res,
+                    metric_name="cpu_pct", operator=">", threshold=90,
+                    consecutive_failures_required=1, consecutive_ok_required=2)
+                rule = await alert_svc.add_rule(s, fields, created_by=user.id)
+            rule_id = rule.id
+
+        async def sample(v):
+            async with app.db_sessionmaker() as s:
+                async with s.begin():
+                    await record_metric(s, "host_agent", res, "cpu_pct", v)
+
+        async def state():
+            async with app.db_sessionmaker() as s:
+                return (await s.execute(select(AlertState).where(
+                    AlertState.rule_id == rule_id))).scalar_one()
+
+        async def events():
+            async with app.db_sessionmaker() as s:
+                rows = (await s.execute(select(AlertEvent).where(AlertEvent.rule_id == rule_id)
+                                        .order_by(AlertEvent.transitioned_at))).scalars().all()
+            return [(e.from_state, e.to_state, e.note) for e in rows]
+
+        await sample(95)
+        assert (await state()).state == AlertStateEnum.firing
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await alert_svc.acknowledge(s, rule_id, None)
+        for _ in range(3):
+            await sample(97)  # still breached: the acknowledgement holds
+        assert (await state()).state == AlertStateEnum.acknowledged
+        await sample(10)      # one OK check of the two it needs
+        assert (await state()).state == AlertStateEnum.acknowledged
+        await sample(10)
+        st = await state()
+        assert st.state == AlertStateEnum.inactive and st.acknowledged_at is None
+        assert await events() == [("inactive", "firing", None), ("acknowledged", "resolved", None)]
+
+        # Changing the threshold of a firing rule resets it, saying why.
+        await sample(95)
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                r = await s.get(AlertRule, rule_id)
+                fields = await alert_svc.rule_fields(
+                    s, name=r.name, source_module=r.source_module, resource_name=r.resource_name,
+                    metric_name=r.metric_name, operator=">", threshold=99,
+                    consecutive_failures_required=1, consecutive_ok_required=2)
+                await alert_svc.apply_rule_change(s, r, fields)
+        assert (await state()).state == AlertStateEnum.inactive
+        assert (await events())[-1] == ("firing", "resolved", "rule changed: threshold")
+
+        # Renaming alone does not reset; disabling does.
+        await sample(100)
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                r = await s.get(AlertRule, rule_id)
+                await alert_svc.apply_rule_change(s, r, {"name": f"renamed {tag}"})
+        assert (await state()).state == AlertStateEnum.firing
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                r = await s.get(AlertRule, rule_id)
+                await alert_svc.apply_rule_change(s, r, {}, enabled=False)
+        assert (await state()).state == AlertStateEnum.inactive
+        assert (await events())[-1] == ("firing", "resolved", "rule disabled")
+
+    asyncio.run(_go())