优化(wecom-public-health-alert-diagnostics-and-debounce): 增强公网健康告警诊断与防抖

scripts/check-public-app-health.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Probe the public OA readiness endpoint and drive a deduplicated alert."""
+"""Probe the public OA readiness endpoint and drive a debounced alert."""
 
 from __future__ import annotations
 
@@ -14,6 +14,7 @@ import stat
 import subprocess
 import sys
 import tempfile
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -23,8 +24,44 @@ from pathlib import Path
 SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$")
 SAFE_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
 MAX_RESPONSE_BYTES = 64 * 1024
+MAX_COUNTER = 1_000_000
+MAX_TIMESTAMP = (1 << 53) - 1
+STATE_SCHEMA_VERSION = 2
+RECOVERY_THRESHOLD = 2
+RECOVERY_WATCH_SECONDS = 60 * 60
+FLAPPING_STABLE_SECONDS = 30 * 60
 HEALTHY_STATUS = "PASS_APP_HEALTH_READY"
 FAILURE_STATUS = "FAIL_APP_HEALTH"
+FLAPPING_STATUS = "FAIL_APP_HEALTH_FLAPPING"
+FAILURE_CATEGORIES = {
+    "dns_resolution_failed",
+    "tcp_connection_failed",
+    "tls_validation_failed",
+    "request_timeout",
+    "http_unavailable",
+    "response_contract_invalid",
+    "response_too_large",
+}
+LEGACY_FAILURE_CATEGORIES = FAILURE_CATEGORIES | {"dns_or_connect_failed"}
+PHASES = {
+    "healthy",
+    "failure_pending",
+    "alerted",
+    "recovery_pending",
+    "recovery_watch",
+    "flapping",
+}
+STATE_KEYS = {
+    "schemaVersion",
+    "phase",
+    "activeCategory",
+    "lastRecoveredCategory",
+    "consecutiveFailures",
+    "consecutiveSuccesses",
+    "recoveredAt",
+    "healthySince",
+    "flapTransitions",
+}
 
 
 class ProbeFailure(Exception):
@@ -89,9 +126,139 @@ def validate_ready_url(value: str) -> str:
     return value
 
 
+def new_probe_state() -> dict[str, object]:
+    return {
+        "schemaVersion": STATE_SCHEMA_VERSION,
+        "phase": "healthy",
+        "activeCategory": None,
+        "lastRecoveredCategory": None,
+        "consecutiveFailures": 0,
+        "consecutiveSuccesses": 0,
+        "recoveredAt": None,
+        "healthySince": None,
+        "flapTransitions": 0,
+    }
+
+
+def is_bounded_int(value: object, minimum: int = 0, maximum: int = MAX_COUNTER) -> bool:
+    return isinstance(value, int) and not isinstance(value, bool) and minimum <= value <= maximum
+
+
+def is_category(value: object, allow_none: bool = True) -> bool:
+    if value is None:
+        return allow_none
+    return isinstance(value, str) and value in LEGACY_FAILURE_CATEGORIES and bool(SAFE_TOKEN.fullmatch(value))
+
+
+def is_timestamp(value: object, allow_none: bool = True) -> bool:
+    if value is None:
+        return allow_none
+    return is_bounded_int(value, maximum=MAX_TIMESTAMP)
+
+
+def validate_v2_state(payload: dict[str, object]) -> bool:
+    if set(payload) != STATE_KEYS or payload.get("schemaVersion") != STATE_SCHEMA_VERSION:
+        return False
+    phase = payload.get("phase")
+    active = payload.get("activeCategory")
+    recovered = payload.get("lastRecoveredCategory")
+    failures = payload.get("consecutiveFailures")
+    successes = payload.get("consecutiveSuccesses")
+    recovered_at = payload.get("recoveredAt")
+    healthy_since = payload.get("healthySince")
+    transitions = payload.get("flapTransitions")
+    if (
+        phase not in PHASES
+        or not is_category(active)
+        or not is_category(recovered)
+        or not is_bounded_int(failures)
+        or not is_bounded_int(successes)
+        or not is_timestamp(recovered_at)
+        or not is_timestamp(healthy_since)
+        or not is_bounded_int(transitions)
+    ):
+        return False
+    if phase == "healthy":
+        return (
+            active is None
+            and recovered is None
+            and failures == 0
+            and successes == 0
+            and recovered_at is None
+            and healthy_since is None
+            and transitions == 0
+        )
+    if phase == "failure_pending":
+        return (
+            active is not None
+            and failures >= 1
+            and successes == 0
+            and healthy_since is None
+            and transitions == 0
+            and ((recovered is None and recovered_at is None) or (recovered is not None and recovered_at is not None))
+        )
+    if phase == "alerted":
+        return (
+            active is not None
+            and failures >= 1
+            and successes == 0
+            and recovered is None
+            and recovered_at is None
+            and healthy_since is None
+            and transitions == 0
+        )
+    if phase == "recovery_pending":
+        return (
+            active is not None
+            and failures == 0
+            and successes >= 1
+            and recovered is None
+            and recovered_at is None
+            and healthy_since is None
+            and transitions == 0
+        )
+    if phase == "recovery_watch":
+        return (
+            active is None
+            and recovered is not None
+            and failures == 0
+            and successes == 0
+            and recovered_at is not None
+            and healthy_since is None
+            and transitions == 0
+        )
+    if phase != "flapping" or active is None or active != recovered or recovered_at is None or transitions < 1:
+        return False
+    if healthy_since is None:
+        return failures >= 1 and successes == 0
+    return failures == 0 and successes >= 1
+
+
+def migrate_v1_state(payload: dict[str, object]) -> dict[str, object] | None:
+    if set(payload) != {"schemaVersion", "consecutiveFailures", "lastCategory"}:
+        return None
+    count = payload.get("consecutiveFailures")
+    category = payload.get("lastCategory")
+    if payload.get("schemaVersion") != 1 or not is_bounded_int(count) or not is_category(category):
+        return None
+    state = new_probe_state()
+    if count == 0:
+        return state if category is None else None
+    if category is None:
+        return None
+    state.update(
+        {
+            "phase": "failure_pending",
+            "activeCategory": category,
+            "consecutiveFailures": count,
+        }
+    )
+    return state
+
+
 def load_probe_state(path: Path) -> dict[str, object]:
     if not path.exists():
-        return {"schemaVersion": 1, "consecutiveFailures": 0, "lastCategory": None}
+        return new_probe_state()
     validate_regular_file(path, "local_state", private=True)
     try:
         payload = json.loads(path.read_text(encoding="utf-8"))
@@ -99,17 +266,14 @@ def load_probe_state(path: Path) -> dict[str, object]:
         fail("local_state_invalid")
     if not isinstance(payload, dict):
         fail("local_state_invalid")
-    count = payload.get("consecutiveFailures")
-    category = payload.get("lastCategory")
-    if (
-        payload.get("schemaVersion") != 1
-        or not isinstance(count, int)
-        or isinstance(count, bool)
-        or count < 0
-        or (category is not None and (not isinstance(category, str) or not SAFE_TOKEN.fullmatch(category)))
-    ):
+    if payload.get("schemaVersion") == STATE_SCHEMA_VERSION:
+        if not validate_v2_state(payload):
+            fail("local_state_invalid")
+        return payload
+    migrated = migrate_v1_state(payload)
+    if migrated is None:
         fail("local_state_invalid")
-    return payload
+    return migrated
 
 
 def atomic_write(path: Path, payload: dict[str, object]) -> None:
@@ -128,13 +292,193 @@ def atomic_write(path: Path, payload: dict[str, object]) -> None:
         raise
 
 
-def classify_url_error(error: urllib.error.URLError) -> str:
-    reason = error.reason
+def capped(value: int, maximum: int = MAX_COUNTER) -> int:
+    return min(value, maximum)
+
+
+def elapsed(now: int, started_at: object) -> int:
+    if not isinstance(started_at, int) or isinstance(started_at, bool):
+        return 0
+    return max(0, now - started_at)
+
+
+def transition_probe_state(
+    state: dict[str, object],
+    category: str | None,
+    now: int,
+    failure_threshold: int,
+) -> tuple[dict[str, object], tuple[str, str] | None]:
+    """Return the next redacted state and an optional notification candidate."""
+    if not validate_v2_state(state) or not is_timestamp(now, allow_none=False):
+        raise ValueError("invalid transition input")
+    if category is not None and category not in FAILURE_CATEGORIES:
+        raise ValueError("invalid transition category")
+    phase = str(state["phase"])
+
+    if category is None:
+        if phase == "healthy":
+            return state.copy(), None
+        if phase == "failure_pending":
+            if state["recoveredAt"] is not None:
+                next_state = state.copy()
+                next_state.update(
+                    {
+                        "phase": "recovery_watch",
+                        "activeCategory": None,
+                        "consecutiveFailures": 0,
+                    }
+                )
+                return next_state, None
+            return new_probe_state(), None
+        if phase == "alerted":
+            next_state = state.copy()
+            next_state.update(
+                {
+                    "phase": "recovery_pending",
+                    "consecutiveFailures": 0,
+                    "consecutiveSuccesses": 1,
+                }
+            )
+            return next_state, None
+        if phase == "recovery_pending":
+            successes = capped(int(state["consecutiveSuccesses"]) + 1)
+            if successes < RECOVERY_THRESHOLD:
+                next_state = state.copy()
+                next_state["consecutiveSuccesses"] = successes
+                return next_state, None
+            recovered_category = str(state["activeCategory"])
+            next_state = new_probe_state()
+            next_state.update(
+                {
+                    "phase": "recovery_watch",
+                    "lastRecoveredCategory": recovered_category,
+                    "recoveredAt": now,
+                }
+            )
+            return next_state, (HEALTHY_STATUS, "public_ready_healthy")
+        if phase == "recovery_watch":
+            if elapsed(now, state["recoveredAt"]) >= RECOVERY_WATCH_SECONDS:
+                return new_probe_state(), None
+            return state.copy(), (HEALTHY_STATUS, "public_ready_healthy")
+        if phase == "flapping":
+            next_state = state.copy()
+            if state["healthySince"] is None:
+                next_state["healthySince"] = now
+                next_state["consecutiveFailures"] = 0
+                next_state["consecutiveSuccesses"] = 1
+                next_state["flapTransitions"] = capped(int(state["flapTransitions"]) + 1)
+            else:
+                next_state["consecutiveSuccesses"] = capped(int(state["consecutiveSuccesses"]) + 1)
+            if elapsed(now, next_state["healthySince"]) >= FLAPPING_STABLE_SECONDS:
+                return new_probe_state(), None
+            return next_state, (FLAPPING_STATUS, str(state["activeCategory"]))
+
+    if phase == "flapping":
+        next_state = state.copy()
+        if state["healthySince"] is not None:
+            next_state["flapTransitions"] = capped(int(state["flapTransitions"]) + 1)
+        next_state.update(
+            {
+                "consecutiveFailures": capped(int(state["consecutiveFailures"]) + 1),
+                "consecutiveSuccesses": 0,
+                "healthySince": None,
+            }
+        )
+        return next_state, (FLAPPING_STATUS, str(state["activeCategory"]))
+
+    if phase == "alerted":
+        next_state = state.copy()
+        next_state.update(
+            {
+                "activeCategory": category,
+                "consecutiveFailures": capped(int(state["consecutiveFailures"]) + 1),
+                "consecutiveSuccesses": 0,
+            }
+        )
+        return next_state, (FAILURE_STATUS, category)
+
+    if phase == "recovery_pending":
+        next_state = new_probe_state()
+        next_state.update(
+            {
+                "phase": "alerted",
+                "activeCategory": category,
+                "consecutiveFailures": 1,
+            }
+        )
+        return next_state, (FAILURE_STATUS, category)
+
+    if phase == "recovery_watch":
+        within_watch = elapsed(now, state["recoveredAt"]) < RECOVERY_WATCH_SECONDS
+        next_state = new_probe_state()
+        next_state.update(
+            {
+                "phase": "failure_pending",
+                "activeCategory": category,
+                "consecutiveFailures": 1,
+                "lastRecoveredCategory": state["lastRecoveredCategory"] if within_watch else None,
+                "recoveredAt": state["recoveredAt"] if within_watch else None,
+            }
+        )
+        return next_state, None
+
+    if phase == "healthy":
+        next_state = new_probe_state()
+        next_state.update(
+            {
+                "phase": "failure_pending",
+                "activeCategory": category,
+                "consecutiveFailures": 1,
+            }
+        )
+        return next_state, None
+
+    failures = capped(int(state["consecutiveFailures"]) + 1)
+    if failures < failure_threshold:
+        next_state = state.copy()
+        next_state.update({"activeCategory": category, "consecutiveFailures": failures})
+        return next_state, None
+    same_recent_failure = (
+        state["lastRecoveredCategory"] == category
+        and state["recoveredAt"] is not None
+        and elapsed(now, state["recoveredAt"]) < RECOVERY_WATCH_SECONDS
+    )
+    if same_recent_failure:
+        next_state = state.copy()
+        next_state.update(
+            {
+                "phase": "flapping",
+                "activeCategory": category,
+                "consecutiveFailures": failures,
+                "consecutiveSuccesses": 0,
+                "healthySince": None,
+                "flapTransitions": 1,
+            }
+        )
+        return next_state, (FLAPPING_STATUS, category)
+    next_state = new_probe_state()
+    next_state.update(
+        {
+            "phase": "alerted",
+            "activeCategory": category,
+            "consecutiveFailures": failures,
+        }
+    )
+    return next_state, (FAILURE_STATUS, category)
+
+
+def classify_connection_error(reason: object) -> str:
     if isinstance(reason, (ssl.SSLError, ssl.CertificateError)):
         return "tls_validation_failed"
     if isinstance(reason, (socket.timeout, TimeoutError)):
         return "request_timeout"
-    return "dns_or_connect_failed"
+    if isinstance(reason, socket.gaierror):
+        return "dns_resolution_failed"
+    return "tcp_connection_failed"
+
+
+def classify_url_error(error: urllib.error.URLError) -> str:
+    return classify_connection_error(error.reason)
 
 
 def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
@@ -166,8 +510,10 @@ def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
         raise ProbeFailure("request_timeout") from None
     except (ssl.SSLError, ssl.CertificateError):
         raise ProbeFailure("tls_validation_failed") from None
+    except socket.gaierror:
+        raise ProbeFailure("dns_resolution_failed") from None
     except (http.client.HTTPException, ConnectionError, OSError):
-        raise ProbeFailure("dns_or_connect_failed") from None
+        raise ProbeFailure("tcp_connection_failed") from None
     if len(body) > MAX_RESPONSE_BYTES:
         raise ProbeFailure("response_too_large")
     try:
@@ -179,7 +525,7 @@ def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
 
 
 def write_summary(path: Path, status: str, category: str) -> None:
-    finding = status == FAILURE_STATUS
+    finding = status in {FAILURE_STATUS, FLAPPING_STATUS}
     payload = {
         "issueId": "APP_HEALTH",
         "status": status,
@@ -217,6 +563,26 @@ def evaluate_alert(alert_helper: Path, webhook_file: Path, summary_file: Path, a
         fail("alert_evaluation_failed")
 
 
+def persist_transition(
+    state_file: Path,
+    summary_file: Path,
+    alert_state_file: Path,
+    state: dict[str, object],
+    candidate: tuple[str, str] | None,
+    alert_helper: Path,
+    webhook_file: Path,
+) -> None:
+    try:
+        atomic_write(state_file, state)
+        if candidate is not None:
+            status, category = candidate
+            write_summary(summary_file, status, category)
+    except OSError:
+        fail("local_state_invalid")
+    if candidate is not None:
+        evaluate_alert(alert_helper, webhook_file, summary_file, alert_state_file)
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("--ready-url", required=True)
@@ -238,45 +604,46 @@ def main() -> None:
     summary_file = args.state_dir / "last-app-health.json"
     alert_state_file = args.state_dir / "app-health-alert-state.json"
     state = load_probe_state(probe_state_file)
+    now = int(time.time())
 
     try:
         probe(ready_url, args.timeout_seconds, args.ca_file)
     except ProbeFailure as error:
-        failures = min(int(state["consecutiveFailures"]) + 1, args.failure_threshold)
         try:
-            atomic_write(
-                probe_state_file,
-                {"schemaVersion": 1, "consecutiveFailures": failures, "lastCategory": error.category},
-            )
-        except OSError:
+            next_state, candidate = transition_probe_state(state, error.category, now, args.failure_threshold)
+        except ValueError:
             fail("local_state_invalid")
-        if failures >= args.failure_threshold:
-            try:
-                write_summary(summary_file, FAILURE_STATUS, error.category)
-            except OSError:
-                fail("local_state_invalid")
-            evaluate_alert(args.alert_helper, args.webhook_file, summary_file, alert_state_file)
-            print(
-                f"[app-health][ERROR] health_check_failed category={error.category} consecutive={failures} alert=evaluated",
-                file=sys.stderr,
-            )
-        else:
-            print(
-                f"[app-health][ERROR] health_check_failed category={error.category} consecutive={failures} alert=pending",
-                file=sys.stderr,
-            )
+        persist_transition(
+            probe_state_file,
+            summary_file,
+            alert_state_file,
+            next_state,
+            candidate,
+            args.alert_helper,
+            args.webhook_file,
+        )
+        alert_result = "evaluated" if candidate is not None else "pending"
+        print(
+            f"[app-health][ERROR] health_check_failed category={error.category} "
+            f"consecutive={next_state['consecutiveFailures']} alert={alert_result}",
+            file=sys.stderr,
+        )
         raise SystemExit(1)
 
     try:
-        atomic_write(
-            probe_state_file,
-            {"schemaVersion": 1, "consecutiveFailures": 0, "lastCategory": None},
-        )
-        write_summary(summary_file, HEALTHY_STATUS, "public_ready_healthy")
-    except OSError:
+        next_state, candidate = transition_probe_state(state, None, now, args.failure_threshold)
+    except ValueError:
         fail("local_state_invalid")
-    evaluate_alert(args.alert_helper, args.webhook_file, summary_file, alert_state_file)
-    print("[app-health][PASS] public readiness is healthy")
+    persist_transition(
+        probe_state_file,
+        summary_file,
+        alert_state_file,
+        next_state,
+        candidate,
+        args.alert_helper,
+        args.webhook_file,
+    )
+    print(f"[app-health][PASS] public readiness is healthy phase={next_state['phase']}")
 
 
 if __name__ == "__main__":

scripts/qa/run-app-health-monitor-contract.sh

@@ -137,6 +137,7 @@ MAIN_CAPTURE="$TEST_DIR/main-capture.jsonl"
 new_state_dir "$MAIN_STATE"
 probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
 [[ "$(capture_lines "$MAIN_CAPTURE")" == "0" ]]
+[[ "$(stat -c %a "$MAIN_STATE/app-health-probe-state.json")" == "600" ]]
 pass "healthy baseline writes state without sending an alert"
 
 set_mode unavailable
@@ -155,14 +156,74 @@ pass "changed failure category sends one updated alert"
 
 set_mode healthy
 probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "2" ]]
+pass "first healthy sample keeps recovery pending"
+probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
 [[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
 probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
 [[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
-pass "recovery sends exactly one notification"
+pass "second healthy sample sends exactly one recovery notification"
+
+set_mode invalid-json
+expect_failure "first same-category recurrence stays pending" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "3" ]]
+expect_failure "second same-category recurrence enters flapping" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
+expect_failure "continued flapping failure remains deduplicated" probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
+set_mode healthy
+probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
+python3 - "$MAIN_STATE/app-health-probe-state.json" <<'PY'
+import json, os, sys, time
+path = sys.argv[1]
+payload = json.load(open(path, encoding="utf-8"))
+assert payload["phase"] == "flapping", payload
+payload["healthySince"] = int(time.time()) - 1800
+with open(path, "w", encoding="utf-8") as handle:
+    json.dump(payload, handle)
+    handle.write("\n")
+os.chmod(path, 0o600)
+PY
+probe "$MAIN_STATE" "$MAIN_CAPTURE" >/dev/null
+[[ "$(capture_lines "$MAIN_CAPTURE")" == "4" ]]
+python3 - "$MAIN_STATE/app-health-probe-state.json" <<'PY'
+import json, sys
+payload = json.load(open(sys.argv[1], encoding="utf-8"))
+assert payload["phase"] == "healthy", payload
+assert payload["flapTransitions"] == 0, payload
+PY
+pass "flapping sends once and exits silently after thirty healthy minutes"
+
+DIFFERENT_STATE="$TEST_DIR/different-state"
+DIFFERENT_CAPTURE="$TEST_DIR/different-capture.jsonl"
+new_state_dir "$DIFFERENT_STATE"
+set_mode healthy
+probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
+set_mode unavailable
+expect_failure "different-case first failure stays pending" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
+expect_failure "different-case second failure alerts" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
+set_mode healthy
+probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
+probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null
+[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "2" ]]
+set_mode invalid-json
+expect_failure "new-category recurrence first sample stays pending" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "2" ]]
+expect_failure "new-category recurrence becomes a normal alert" probe "$DIFFERENT_STATE" "$DIFFERENT_CAPTURE" >/dev/null 2>&1
+[[ "$(capture_lines "$DIFFERENT_CAPTURE")" == "3" ]]
+python3 - "$DIFFERENT_STATE/app-health-probe-state.json" <<'PY'
+import json, sys
+payload = json.load(open(sys.argv[1], encoding="utf-8"))
+assert payload["phase"] == "alerted", payload
+assert payload["activeCategory"] == "response_contract_invalid", payload
+PY
+pass "different failure category is not swallowed by same-category flapping debounce"
 
 TRANSIENT_STATE="$TEST_DIR/transient-state"
 TRANSIENT_CAPTURE="$TEST_DIR/transient-capture.jsonl"
 new_state_dir "$TRANSIENT_STATE"
+set_mode healthy
 probe "$TRANSIENT_STATE" "$TRANSIENT_CAPTURE" >/dev/null
 set_mode unavailable
 expect_failure "single transient outage is recorded" probe "$TRANSIENT_STATE" "$TRANSIENT_CAPTURE" >/dev/null 2>&1
@@ -184,10 +245,48 @@ expected = {
     "timeout": "request_timeout",
 }[sys.argv[2]]
 payload = json.load(open(sys.argv[1], encoding="utf-8"))
-assert payload["lastCategory"] == expected, payload
+assert payload["activeCategory"] == expected, payload
 PY
 done
 
+python3 - "$ROOT_DIR/scripts/check-public-app-health.py" <<'PY'
+import importlib.util
+import socket
+import sys
+import urllib.error
+
+spec = importlib.util.spec_from_file_location("app_health", sys.argv[1])
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+assert module.classify_url_error(urllib.error.URLError(socket.gaierror(-2, "private-dns-text"))) == "dns_resolution_failed"
+assert module.classify_url_error(urllib.error.URLError(ConnectionRefusedError())) == "tcp_connection_failed"
+
+watch = module.new_probe_state()
+watch.update({
+    "phase": "recovery_watch",
+    "lastRecoveredCategory": "http_unavailable",
+    "recoveredAt": 1000,
+})
+pending, candidate = module.transition_probe_state(watch, "http_unavailable", 4599, 2)
+assert pending["phase"] == "failure_pending" and candidate is None, (pending, candidate)
+flapping, candidate = module.transition_probe_state(pending, "http_unavailable", 4599, 2)
+assert flapping["phase"] == "flapping", flapping
+assert candidate == ("FAIL_APP_HEALTH_FLAPPING", "http_unavailable"), candidate
+
+expired, candidate = module.transition_probe_state(watch, "http_unavailable", 4600, 2)
+assert expired["phase"] == "failure_pending" and expired["recoveredAt"] is None, expired
+alerted, candidate = module.transition_probe_state(expired, "http_unavailable", 4600, 2)
+assert alerted["phase"] == "alerted", alerted
+assert candidate == ("FAIL_APP_HEALTH", "http_unavailable"), candidate
+
+flapping["healthySince"] = 5000
+flapping["consecutiveFailures"] = 0
+flapping["consecutiveSuccesses"] = 1
+stable, candidate = module.transition_probe_state(flapping, None, 6800, 2)
+assert stable["phase"] == "healthy" and candidate is None, (stable, candidate)
+PY
+pass "DNS/TCP classification and exact thirty/sixty-minute boundaries are deterministic"
+
 UNUSED_PORT="$(python3 - <<'PY'
 import socket
 sock = socket.socket()
@@ -207,9 +306,9 @@ expect_failure "connection failure is rejected" env \
 python3 - "$CONNECT_STATE/app-health-probe-state.json" <<'PY'
 import json, sys
 payload = json.load(open(sys.argv[1], encoding="utf-8"))
-assert payload["lastCategory"] == "dns_or_connect_failed", payload
+assert payload["activeCategory"] == "tcp_connection_failed", payload
 PY
-pass "connection failure is stored as a stable category"
+pass "TCP connection failure is stored separately from DNS resolution failure"
 
 TLS_STATE="$TEST_DIR/tls-state"
 new_state_dir "$TLS_STATE"
@@ -222,10 +321,39 @@ expect_failure "untrusted TLS certificate is rejected" env \
 python3 - "$TLS_STATE/app-health-probe-state.json" <<'PY'
 import json, sys
 payload = json.load(open(sys.argv[1], encoding="utf-8"))
-assert payload["lastCategory"] == "tls_validation_failed", payload
+assert payload["activeCategory"] == "tls_validation_failed", payload
 PY
 pass "TLS failure is stored as a stable category"
 
+MIGRATION_STATE="$TEST_DIR/migration-state"
+MIGRATION_CAPTURE="$TEST_DIR/migration-capture.jsonl"
+new_state_dir "$MIGRATION_STATE"
+printf '%s\n' '{"schemaVersion":1,"consecutiveFailures":0,"lastCategory":null}' \
+  >"$MIGRATION_STATE/app-health-probe-state.json"
+chmod 0600 "$MIGRATION_STATE/app-health-probe-state.json"
+set_mode healthy
+probe "$MIGRATION_STATE" "$MIGRATION_CAPTURE" >/dev/null
+python3 - "$MIGRATION_STATE/app-health-probe-state.json" <<'PY'
+import json, sys
+payload = json.load(open(sys.argv[1], encoding="utf-8"))
+assert payload["schemaVersion"] == 2, payload
+assert payload["phase"] == "healthy", payload
+PY
+[[ "$(capture_lines "$MIGRATION_CAPTURE")" == "0" ]]
+pass "valid legacy state migrates without backfilling a notification"
+
+INVALID_SCHEMA_STATE="$TEST_DIR/invalid-schema-state"
+new_state_dir "$INVALID_SCHEMA_STATE"
+printf '%s\n' '{"schemaVersion":2,"phase":"healthy","activeCategory":null,"lastRecoveredCategory":null,"consecutiveFailures":1000001,"consecutiveSuccesses":0,"recoveredAt":null,"healthySince":null,"flapTransitions":0}' \
+  >"$INVALID_SCHEMA_STATE/app-health-probe-state.json"
+chmod 0600 "$INVALID_SCHEMA_STATE/app-health-probe-state.json"
+set_mode healthy
+expect_failure "out-of-range v2 state fails closed" probe "$INVALID_SCHEMA_STATE" \
+  "$TEST_DIR/invalid-schema-capture.jsonl" >"$TEST_DIR/invalid-schema-stdout" 2>"$TEST_DIR/invalid-schema-stderr"
+grep -Fxq '[app-health][ERROR] local_state_invalid' "$TEST_DIR/invalid-schema-stderr"
+! grep -Fq '1000001' "$TEST_DIR/invalid-schema-stderr"
+pass "v2 state schema rejects inconsistent or out-of-range fields without disclosure"
+
 CORRUPT_STATE="$TEST_DIR/corrupt-state"
 new_state_dir "$CORRUPT_STATE"
 printf '%s\n' 'private-payload-must-not-appear' >"$CORRUPT_STATE/app-health-probe-state.json"
@@ -241,15 +369,19 @@ pass "local state error output is stable and redacted"
 python3 - "$MAIN_CAPTURE" <<'PY'
 import json, sys
 messages = [json.loads(line)["text"]["content"] for line in open(sys.argv[1], encoding="utf-8")]
-assert len(messages) == 3, messages
+assert len(messages) == 4, messages
 assert "FAIL_APP_HEALTH" in messages[0]
 assert "http_unavailable" in messages[0]
 assert "response_contract_invalid" in messages[1]
 assert "PASS_APP_HEALTH_READY" in messages[2]
+assert "FAIL_APP_HEALTH_FLAPPING" in messages[3]
+assert messages[0].startswith("OA 公网健康告警\n"), messages[0]
+assert messages[2].startswith("OA 公网健康恢复\n"), messages[2]
+assert messages[3].startswith("OA 公网健康链路反复抖动\n"), messages[3]
 for forbidden in ("do-not-leak", "localhost", "https://", "/tmp/", "not-json-private-payload"):
     assert all(forbidden not in message for message in messages), (forbidden, messages)
 PY
-pass "captured alerts contain only stable status and category text"
+pass "captured application alerts use explicit source titles and stable redacted categories"
 
 expect_failure "plain HTTP readiness URL is rejected" env \
   HYB_WECOM_ALERT_TEST_MODE=1 HYB_WECOM_ALERT_TEST_CAPTURE="$TEST_DIR/http-capture.jsonl" \

scripts/qa/run-automatic-backup-contract.sh

@@ -701,6 +701,16 @@ mv "$TEST_DIR/deadman-withheld-batch" "$DEADMAN_LAYOUT/archive/$SECOND_BATCH"
   --webhook-file "$TEST_DIR/wecom-webhook" --alert-helper "$ROOT_DIR/scripts/send-wecom-group-alert.py" >/dev/null
 [[ "$(wc -l <"$DEADMAN_CAPTURE")" == "2" ]]
 pass "ARM dead-man sends one recovery notification after finalized archives resume"
+python3 - "$DEADMAN_CAPTURE" <<'PYTHON'
+import json
+import sys
+
+messages = [json.loads(line)["text"]["content"] for line in open(sys.argv[1], encoding="utf-8")]
+assert messages[0].startswith("内部 OA 运维告警\n"), messages
+assert messages[1].startswith("内部 OA 运维恢复通知\n"), messages
+assert all("OA 公网健康" not in message for message in messages), messages
+PYTHON
+pass "shared helper keeps dead-man alert and recovery titles unchanged"
 
 DRIFT_BUILD_SHA="0123456789abcdef0123456789abcdef01234567"
 expect_failure "wrapper rejects an active runtime build that drifted from its installed contract" \

scripts/send-wecom-group-alert.py

@@ -18,6 +18,7 @@ from pathlib import Path
 SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$")
 ALLOWED_STATUSES = {
     "FAIL_APP_HEALTH",
+    "FAIL_APP_HEALTH_FLAPPING",
     "FAIL_AUTOMATIC_BACKUP",
     "FAIL_OFFHOST_DEADMAN",
     "PASS_APP_HEALTH_READY",
@@ -131,7 +132,13 @@ def should_send(previous: dict[str, object], status: str, category: str) -> bool
 
 
 def safe_text(status: str, category: str) -> str:
-    if status.startswith("FAIL_"):
+    if status == "FAIL_APP_HEALTH":
+        headline = "OA 公网健康告警"
+    elif status == "FAIL_APP_HEALTH_FLAPPING":
+        headline = "OA 公网健康链路反复抖动"
+    elif status == "PASS_APP_HEALTH_READY":
+        headline = "OA 公网健康恢复"
+    elif status.startswith("FAIL_"):
         headline = "内部 OA 运维告警"
     elif status == "PASS_WITH_FINDINGS":
         headline = "内部 OA 运维提醒"