feat(runtime-guard): NOTIFY-04 B5 alert de-duplication (dry_run by default)

.github/workflows/runtime-guard.yml

@@ -59,6 +59,10 @@ jobs:
       GLOBAL_TELEGRAM_CHAT_ID: ${{ secrets.GLOBAL_TELEGRAM_CHAT_ID }}
       TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
       TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }}
+      # NOTIFY-04 B5: default dry_run = log would_suppress only, never suppress.
+      RUNTIME_GUARD_ALERT_DEDUPE_MODE: ${{ vars.RUNTIME_GUARD_ALERT_DEDUPE_MODE || 'dry_run' }}
+      RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS: ${{ vars.RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS || '6' }}
+      RUNTIME_GUARD_DEDUPE_STATE_PATH: .runtime-guard-dedupe/state.json
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6
@@ -98,5 +102,20 @@ jobs:
       - name: Set up gcloud
         uses: google-github-actions/setup-gcloud@v3
 
+      - name: Restore runtime-guard alert dedupe state
+        uses: actions/cache/restore@v4
+        with:
+          path: .runtime-guard-dedupe
+          key: runtime-guard-dedupe-${{ github.run_id }}-${{ github.run_attempt }}
+          restore-keys: |
+            runtime-guard-dedupe-
+
       - name: Check Cloud Scheduler and Cloud Run logs
         run: uv run --no-sync python scripts/cloud_run_runtime_guard.py
+
+      - name: Save runtime-guard alert dedupe state
+        if: ${{ always() && hashFiles('.runtime-guard-dedupe/state.json') != '' }}
+        uses: actions/cache/save@v4
+        with:
+          path: .runtime-guard-dedupe
+          key: runtime-guard-dedupe-${{ github.run_id }}-${{ github.run_attempt }}

scripts/cloud_run_runtime_guard.py

@@ -20,6 +20,11 @@
     resolve_operational_notification_locale,
 )
 
+try:
+    from scripts import runtime_guard_alert_dedupe as alert_dedupe
+except ModuleNotFoundError:  # executed as `python scripts/cloud_run_runtime_guard.py`
+    import runtime_guard_alert_dedupe as alert_dedupe  # type: ignore[no-redef]
+
 
 ERROR_SEVERITIES = {"ERROR", "CRITICAL", "ALERT", "EMERGENCY"}
 FAILURE_WORDS = (
@@ -673,6 +678,7 @@ def main() -> int:
 
     issues: list[str] = []
     details: list[str] = []
+    alert_event_ids: list[str] = []
     success_count = 0
     success_count_by_service: dict[str, int] = {}
     queried_services: set[str] = set()
@@ -716,6 +722,7 @@ def main() -> int:
                 )
             )
             details.extend(_summarize(entry) for entry in failures[:5])
+            alert_event_ids.extend(alert_dedupe.event_id(entry) for entry in failures)
 
     if services and require_success:
         for service in _services_without_success(
@@ -764,13 +771,15 @@ def main() -> int:
                     )
                 )
                 details.extend(_summarize(entry) for entry in failures[:5])
+                alert_event_ids.extend(alert_dedupe.event_id(entry) for entry in failures)
         except RuntimeError as exc:
             issues.append(_notice("runtime_guard_scheduler_log_query_failed"))
             details.append(f"Cloud Scheduler log query failed: {exc}")
     elif check_scheduler:
         print("Skipping Cloud Scheduler check because no scheduler job pattern could be derived.", file=sys.stderr)
 
     if not issues:
+        alert_dedupe.clear_issues()
         service_text = ", ".join(services) if services else "<none configured>"
         print(
             f"Runtime guard OK for {name}: services={service_text}, lookback={lookback_minutes}m, successes={success_count}"
@@ -793,6 +802,11 @@ def main() -> int:
         workflow_url=run_url,
     )
     print(message)
+    decision = alert_dedupe.decide(name=name, issues=issues, event_ids=alert_event_ids)
+    print(decision.log_line())
+    if not decision.send:
+        print(f"Runtime guard alert suppressed as a repeat for {name}; no new failure entries.")
+        return 1 if fail_workflow else 0
     _send_telegram(message[:3900])
     return 1 if fail_workflow else 0
 

scripts/runtime_guard_alert_dedupe.py

@@ -0,0 +1,202 @@
+"""Runtime-guard alert de-duplication (NOTIFY-04 B5).
+
+The guard runs every 30 minutes with a 180-minute log lookback, so one Cloud
+Run / Cloud Scheduler failure entry is re-reported by up to six consecutive
+runs. This module decides whether an alert is a pure repeat:
+
+* new failure log entries (by event id) → always alert;
+* a new issue type (normalized issue text fingerprint) → always alert;
+* the same issues with no new entries → repeat; re-alert after
+  ``RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS`` (default 6).
+
+Modes (``RUNTIME_GUARD_ALERT_DEDUPE_MODE``):
+
+* ``off`` (default when unset): no state, behaviour unchanged;
+* ``dry_run``: never suppresses; logs ``would_suppress`` and keeps state as if
+  ``on`` were active, so the effect can be observed before switching;
+* ``on``: suppresses pure repeats.
+
+State is a small JSON file (hashes and timestamps only, no log text, no
+account data) persisted between runs by the workflow cache step.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import hashlib
+import json
+import os
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterable, Mapping, Sequence
+
+MODES = ("off", "dry_run", "on")
+DEFAULT_STATE_PATH = ".runtime-guard-dedupe/state.json"
+DEFAULT_REMINDER_HOURS = 6.0
+EVENT_RETENTION_HOURS = 48.0
+STATE_VERSION = 1
+
+
+@dataclass(frozen=True)
+class DedupeDecision:
+    mode: str
+    send: bool
+    would_suppress: bool
+    new_event_count: int
+    new_issue_fingerprint: bool
+    reason: str
+
+    def log_line(self) -> str:
+        return (
+            "runtime_guard_dedupe "
+            f"mode={self.mode} send={str(self.send).lower()} "
+            f"would_suppress={str(self.would_suppress).lower()} "
+            f"new_events={self.new_event_count} "
+            f"new_issue_type={str(self.new_issue_fingerprint).lower()} reason={self.reason}"
+        )
+
+
+def resolve_mode(raw: str | None = None) -> str:
+    value = str(raw if raw is not None else os.environ.get("RUNTIME_GUARD_ALERT_DEDUPE_MODE") or "").strip().lower()
+    return value if value in MODES else "off"
+
+
+def _state_path(raw: str | None = None) -> Path:
+    return Path(raw or os.environ.get("RUNTIME_GUARD_DEDUPE_STATE_PATH") or DEFAULT_STATE_PATH)
+
+
+def _reminder_hours() -> float:
+    try:
+        value = float(os.environ.get("RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS") or DEFAULT_REMINDER_HOURS)
+    except ValueError:
+        return DEFAULT_REMINDER_HOURS
+    return value if value > 0 else DEFAULT_REMINDER_HOURS
+
+
+def event_id(entry: Mapping[str, Any]) -> str:
+    """Stable id for one log entry (Cloud Logging insertId preferred)."""
+
+    insert_id = str(entry.get("insertId") or "").strip()
+    if insert_id:
+        basis = f"insert:{insert_id}"
+    else:
+        resource = entry.get("resource") if isinstance(entry.get("resource"), Mapping) else {}
+        basis = json.dumps(
+            {
+                "timestamp": entry.get("timestamp"),
+                "logName": entry.get("logName"),
+                "resource": resource.get("labels") if isinstance(resource, Mapping) else None,
+                "severity": entry.get("severity"),
+            },
+            sort_keys=True,
+            default=str,
+        )
+    return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:24]
+
+
+def issue_fingerprint(name: str, issues: Sequence[str]) -> str:
+    """Issue-type fingerprint; counts/minutes are normalized so N→N+1 is the same type."""
+
+    normalized = sorted({re.sub(r"\d+", "#", str(item)).strip() for item in issues})
+    basis = json.dumps({"name": name, "issues": normalized}, ensure_ascii=False)
+    return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:24]
+
+
+def _parse(ts: Any) -> dt.datetime | None:
+    try:
+        value = dt.datetime.fromisoformat(str(ts))
+    except (TypeError, ValueError):
+        return None
+    return value if value.tzinfo else value.replace(tzinfo=dt.timezone.utc)
+
+
+def _load(path: Path) -> dict[str, Any]:
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return {"version": STATE_VERSION, "events": {}, "issues": {}}
+    if not isinstance(payload, dict) or payload.get("version") != STATE_VERSION:
+        return {"version": STATE_VERSION, "events": {}, "issues": {}}
+    payload.setdefault("events", {})
+    payload.setdefault("issues", {})
+    return payload
+
+
+def _save(path: Path, state: Mapping[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(state, sort_keys=True), encoding="utf-8")
+
+
+def _prune(state: dict[str, Any], now: dt.datetime) -> None:
+    horizon = now - dt.timedelta(hours=EVENT_RETENTION_HOURS)
+    state["events"] = {
+        key: value
+        for key, value in dict(state.get("events") or {}).items()
+        if (_parse(value) or now) >= horizon
+    }
+
+
+def decide(
+    *,
+    name: str,
+    issues: Sequence[str],
+    event_ids: Iterable[str],
+    now: dt.datetime | None = None,
+    mode: str | None = None,
+    state_path: str | None = None,
+) -> DedupeDecision:
+    """Decide whether to send this alert and record it in the state."""
+
+    resolved = resolve_mode(mode)
+    ids = sorted({str(item) for item in event_ids if str(item)})
+    if resolved == "off":
+        return DedupeDecision(resolved, True, False, len(ids), True, "dedupe_off")
+    stamp = now or dt.datetime.now(dt.timezone.utc)
+    path = _state_path(state_path)
+    state = _load(path)
+    _prune(state, stamp)
+    events: dict[str, str] = state["events"]
+    issues_state: dict[str, str] = state["issues"]
+
+    new_ids = [item for item in ids if item not in events]
+    fingerprint = issue_fingerprint(name, issues)
+    last_alert = _parse(issues_state.get(fingerprint))
+    new_type = last_alert is None
+    reminder_due = (
+        last_alert is not None
+        and stamp - last_alert >= dt.timedelta(hours=_reminder_hours())
+    )
+    if new_ids:
+        repeat, reason = False, "new_failure_events"
+    elif new_type:
+        repeat, reason = False, "new_issue_type"
+    elif reminder_due:
+        repeat, reason = False, "reminder_due"
+    else:
+        repeat, reason = True, "repeat_within_reminder_window"
+
+    if not repeat:
+        for item in ids:
+            events.setdefault(item, stamp.isoformat())
+        issues_state[fingerprint] = stamp.isoformat()
+    else:
+        for item in ids:
+            events.setdefault(item, stamp.isoformat())
+    _save(path, state)
+
+    send = not (repeat and resolved == "on")
+    return DedupeDecision(resolved, send, repeat, len(new_ids), new_type, reason)
+
+
+def clear_issues(*, mode: str | None = None, state_path: str | None = None, now: dt.datetime | None = None) -> None:
+    """Healthy run: forget issue fingerprints so a recurrence alerts immediately."""
+
+    if resolve_mode(mode) == "off":
+        return
+    stamp = now or dt.datetime.now(dt.timezone.utc)
+    path = _state_path(state_path)
+    state = _load(path)
+    _prune(state, stamp)
+    state["issues"] = {}
+    _save(path, state)