feat: dedup alerts so one failure = one incident

core/main.py

@@ -27,7 +27,8 @@ async def lifespan(app: FastAPI):
 @app.post("/alert")
 def receive_alert(payload: dict, background_tasks: BackgroundTasks) -> dict:
     incident = orchestrator.handle_alert(payload)
-    background_tasks.add_task(orchestrator.run_diagnostics, incident["incident_id"], payload)
+    if not incident.get("suppressed"):
+        background_tasks.add_task(orchestrator.run_diagnostics, incident["incident_id"], payload)
     return incident
 
 

core/orchestrator.py

@@ -1,23 +1,61 @@
 import os
 import time
+from collections import defaultdict
 from datetime import datetime, timezone
 from . import db
 from .services import git_client, llm_analyzer, vector_store, notifier, postmortem
 
 REPO_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))  # Sentinel root
 RUNBOOKS_DIR = os.path.join(REPO_PATH, "sandbox", "runbooks")
 
+# ponytail: in-process sliding-window dedup; per-worker, resets on restart.
+# Move to Postgres/Redis if you run multiple workers or need durability.
+_DEDUP_WINDOW_S = float(os.getenv("SENTINEL_DEDUP_WINDOW_S", "300"))
+_DEDUP_THRESHOLD = int(os.getenv("SENTINEL_DEDUP_THRESHOLD", "1"))  # Nth hit in window fires
+_recent_alerts: dict[str, list[float]] = defaultdict(list)          # signature -> hit times
+_open_signatures: dict[str, str] = {}                              # signature -> live incident_id
+
+
+def _dedup_gate(signature: str) -> tuple[str, str | None]:
+    """'fire' (open a new incident), 'duplicate' (fold into an active one),
+    or 'suppress' (below the burst threshold)."""
+    now = time.monotonic()
+    hits = _recent_alerts[signature]
+    hits[:] = [t for t in hits if t >= now - _DEDUP_WINDOW_S]
+    hits.append(now)
+
+    if signature in _open_signatures:
+        return "duplicate", _open_signatures[signature]
+    if len(hits) < _DEDUP_THRESHOLD:
+        return "suppress", None
+    return "fire", None
+
+
+def _clear_signature(signature: str):
+    _open_signatures.pop(signature, None)
+
 
 def handle_alert(alert_data: dict) -> dict:
+    signature = alert_data.get("error_signature", "")
+    decision, existing_id = _dedup_gate(signature)
+    if decision == "duplicate":
+        print(f"[dedup] alert folded into active incident {existing_id}")
+        return {"incident_id": existing_id, "status": "duplicate", "suppressed": True}
+    if decision == "suppress":
+        seen = len(_recent_alerts[signature])
+        print(f"[dedup] alert suppressed ({seen}/{_DEDUP_THRESHOLD} in {int(_DEDUP_WINDOW_S)}s window)")
+        return {"incident_id": None, "status": "suppressed", "suppressed": True}
+
     ts = datetime.now(timezone.utc).strftime("%Y_%m%d_%H%M%S")
     incident = {
-        "incident_id": f"inc_{ts}",
+        "incident_id": f"inc_{ts}_{os.urandom(2).hex()}",  # suffix: same-second alerts must not collide on PK
         "status": "triggered",
         "trigger": alert_data,
         "diagnostics": None,
         "postmortem_draft_url": None,
     }
     db.save_incident(incident)
+    _open_signatures[signature] = incident["incident_id"]
     print(f"[orchestrator] {incident['incident_id']} created -> triggered")
     return incident
 
@@ -61,6 +99,7 @@ def run_diagnostics(incident_id: str, alert_data: dict):
         db.update_diagnostics(incident_id, diagnostics)
         if degraded_reason:
             db.update_status(incident_id, "degraded")
+            _clear_signature(alert_data.get("error_signature", ""))  # terminal: stop folding new alerts
             print(f"[orchestrator] {incident_id} -> degraded (terminal)")
         else:
             alert_dt = datetime.fromisoformat(alert_data["timestamp"].replace("Z", "+00:00"))
@@ -76,6 +115,7 @@ def run_diagnostics(incident_id: str, alert_data: dict):
     except Exception as e:
         print(f"[orchestrator] {incident_id} diagnostics failed: {e}")
         db.update_status(incident_id, "triggered")  # revert so it can be retried
+        _clear_signature(alert_data.get("error_signature", ""))  # allow a retry alert through
 
 
 def resolve_incident(incident_id: str) -> dict:
@@ -92,6 +132,7 @@ def resolve_incident(incident_id: str) -> dict:
         )
         print(f"[orchestrator] {incident_id} -- postmortem LLM unavailable, degrading: {e}")
     db.update_postmortem(incident_id, draft)
+    _clear_signature((incident.get("trigger_data") or {}).get("error_signature", ""))  # resolved: reopen to new alerts
     print(f"[orchestrator] {incident_id} -> resolved, postmortem generated")
     print(f"[metrics] {incident_id} resolve_to_postmortem_s={time.monotonic() - t0:.1f}")
     return db.get_incident(incident_id)

core/tests/test_orchestrator.py

@@ -1,5 +1,6 @@
 # ponytail: mocks db so test runs without a live postgres connection
 from unittest.mock import patch
+import core.orchestrator as orch
 from core.orchestrator import handle_alert
 
 _ALERT = {
@@ -10,7 +11,13 @@
 }
 
 
+def _reset_dedup():
+    orch._recent_alerts.clear()
+    orch._open_signatures.clear()
+
+
 def test_handle_alert():
+    _reset_dedup()
     with patch("core.orchestrator.db.save_incident") as mock_save:
         result = handle_alert(_ALERT)
     assert result["status"] == "triggered"
@@ -20,6 +27,41 @@ def test_handle_alert():
     mock_save.assert_called_once()
 
 
+def test_dedup_folds_repeat_alerts():
+    """Second alert for a still-open signature must not open a new incident."""
+    _reset_dedup()
+    with patch("core.orchestrator.db.save_incident") as mock_save:
+        first = handle_alert(_ALERT)
+        second = handle_alert(_ALERT)
+    assert first["status"] == "triggered"
+    assert second["status"] == "duplicate"
+    assert second["incident_id"] == first["incident_id"]
+    mock_save.assert_called_once()  # only the first alert hit the DB
+
+    # once the incident is cleared (resolved/degraded/failed), a new alert fires again
+    orch._clear_signature(_ALERT["error_signature"])
+    with patch("core.orchestrator.db.save_incident"):
+        third = handle_alert(_ALERT)
+    assert third["status"] == "triggered"
+    assert third["incident_id"] != first["incident_id"]
+
+
+def test_dedup_threshold_suppresses_below_burst():
+    """With a threshold >1, alerts below the burst count are suppressed, not opened."""
+    _reset_dedup()
+    with patch.object(orch, "_DEDUP_THRESHOLD", 3), \
+         patch("core.orchestrator.db.save_incident") as mock_save:
+        r1 = handle_alert(_ALERT)
+        r2 = handle_alert(_ALERT)
+        r3 = handle_alert(_ALERT)
+    assert r1["status"] == "suppressed"
+    assert r2["status"] == "suppressed"
+    assert r3["status"] == "triggered"  # 3rd hit crosses the threshold
+    mock_save.assert_called_once()
+
+
 if __name__ == "__main__":
     test_handle_alert()
-    print("✓ orchestrator test passed")
+    test_dedup_folds_repeat_alerts()
+    test_dedup_threshold_suppresses_below_burst()
+    print("[ok] orchestrator + dedup tests passed")