fix: harden runtime deployment defaults

.gitignore

@@ -17,6 +17,8 @@ dist/
 # Env
 .env
 backend/.env
+backend/.env.*
+!backend/.env.example
 
 # Database
 *.db

backend/.env.example

@@ -1,20 +1,33 @@
+# 运行环境
+ENVIRONMENT=development
+
 # 数据库（开发用 SQLite，生产用 PostgreSQL）
 DATABASE_URL=sqlite+aiosqlite:///./pingwatch.db
-# DATABASE_URL=postgresql+asyncpg://user:pass@localhost/pingwatch
 
 # 企业微信
-WECOM_CORP_ID=your_corp_id
+WECOM_NOTIFICATION_ENABLED=false
-WECOM_AGENT_ID=1000001
+WECOM_CORP_ID=
-WECOM_APP_SECRET=your_app_secret
+WECOM_AGENT_ID=0
+WECOM_APP_SECRET=
+WECOM_TO_PARTY=
+WECOM_NOTIFICATION_MAX_ATTEMPTS=5
+WECOM_RETRY_BASE_SECONDS=30
 
 # Casdoor
 CASDOOR_ENDPOINT=https://casdoor.dhdx.fun
-CASDOOR_CLIENT_ID=e46b9e1eb893027bdf2a
+CASDOOR_CLIENT_ID=
-CASDOOR_CLIENT_SECRET=b12c7e1688ed51481f3b5c5dae4191b6edbba916
+CASDOOR_CLIENT_SECRET=
 CASDOOR_CERTIFICATE=
 CASDOOR_ORGANIZATION=dahua
 CASDOOR_APPLICATION=PingWatch
-CASDOOR_REDIRECT_URI=http://10.10.10.7:5173/login
+CASDOOR_REDIRECT_URI=
+
+# 连通性监测
+PROBE_PACKETS_PER_ROUND=3
+OFFLINE_CONSECUTIVE_ROUNDS=2
+DEGRADED_WINDOW_ROUNDS=5
+DEGRADED_LOSS_PERCENT=20.0
+RECOVERY_CONSECUTIVE_CLEAN_ROUNDS=3
 
 # Ping 引擎
 PING_INTERVAL_SECONDS=30
@@ -31,5 +44,8 @@ OFFLINE_SUPPRESS_RATIO=0.9
 PING_RECORD_RETENTION_DAYS=90
 ALERT_RETENTION_DAYS=365
 
-# JWT 密钥（请改成随机字符串）
+# JWT 密钥（生产环境必须由受控运行环境注入）
-SECRET_KEY=change-me-to-a-long-random-string
+SECRET_KEY=
+
+# LogHive（可选）
+LOGHIVE_API_KEY=

backend/app/config.py

@@ -1,10 +1,17 @@
-"""应用配置，通过环境变量注入，不支持 .env 文件"""
+"""Application configuration loaded from environment variables or a local .env file."""
 
+from pydantic import Field
 from pydantic_settings import BaseSettings
-from typing import Optional
+from typing import Literal
+
+
+LEGACY_DEFAULT_SECRET = "change-me-to-a-long-random-string"
 
 
 class Settings(BaseSettings):
+    # ---------- 运行环境 ----------
+    environment: Literal["development", "test", "production"] = "development"
+
     # ---------- 数据库 ----------
     DATABASE_URL: str = "sqlite+aiosqlite:///./pingwatch.db"
     # PostgreSQL: "postgresql+asyncpg://user:pass@localhost/pingwatch"
@@ -15,7 +22,7 @@ class Settings(BaseSettings):
     WECOM_APP_SECRET: str = ""
 
     # ---------- Casdoor ----------
-    CASDOOR_ENDPOINT: str = "https://casdoor.dhdx.fun"
+    casdoor_endpoint: str = "https://casdoor.dhdx.fun"
     CASDOOR_CLIENT_ID: str = ""
     CASDOOR_CLIENT_SECRET: str = ""
     CASDOOR_CERTIFICATE: str = ""               # 可选，用于验证 id_token 签名
@@ -23,6 +30,19 @@ class Settings(BaseSettings):
     CASDOOR_APPLICATION: str = "PingWatch"
     CASDOOR_REDIRECT_URI: str = "http://10.10.10.7:5173/login"  # 前端回调地址
 
+    # ---------- 连通性监测 ----------
+    probe_packets_per_round: int = Field(default=3, ge=1, le=10)
+    offline_consecutive_rounds: int = Field(default=2, ge=1, le=10)
+    degraded_window_rounds: int = Field(default=5, ge=2, le=60)
+    degraded_loss_percent: float = Field(default=20.0, ge=1, le=100)
+    recovery_consecutive_clean_rounds: int = Field(default=3, ge=1, le=20)
+
+    # ---------- 企业微信投递 ----------
+    wecom_notification_enabled: bool = False
+    wecom_notification_max_attempts: int = Field(default=5, ge=1, le=10)
+    wecom_retry_base_seconds: int = Field(default=30, ge=1, le=3600)
+    WECOM_TO_PARTY: str = ""
+
     # ---------- Ping 引擎 ----------
     PING_INTERVAL_SECONDS: int = 30
     PING_TIMEOUT_SECONDS: float = 5.0
@@ -40,7 +60,7 @@ class Settings(BaseSettings):
     ALERT_RETENTION_DAYS: int = 365
 
     # ---------- JWT ----------
-    SECRET_KEY: str = "change-me-to-a-long-random-string"
+    secret_key: str = LEGACY_DEFAULT_SECRET
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 480
 
     # ---------- LogHive ----------
@@ -53,5 +73,15 @@ class Settings(BaseSettings):
 
     model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
 
+    @property
+    def CASDOOR_ENDPOINT(self) -> str:
+        """Compatibility accessor for existing uppercase configuration consumers."""
+        return self.casdoor_endpoint
+
+    @property
+    def SECRET_KEY(self) -> str:
+        """Compatibility accessor for existing uppercase configuration consumers."""
+        return self.secret_key
+
 
 settings = Settings()

backend/app/core/auth.py

@@ -64,11 +64,11 @@ async def exchange_code_for_user(code: str) -> Optional[dict]:
         "code": code,
     }
 
-    async with httpx.AsyncClient(timeout=15, verify=False) as client:
+    async with httpx.AsyncClient(timeout=15) as client:
         try:
             resp = await client.post(token_url, data=data)
             if resp.status_code != 200:
-                logger.error(f"Casdoor token 换取失败: {resp.status_code} {resp.text}")
+                logger.error("Casdoor token 换取失败: status=%s", resp.status_code)
                 return None
 
             token_data = resp.json()
@@ -77,19 +77,16 @@ async def exchange_code_for_user(code: str) -> Optional[dict]:
                 logger.error("Casdoor 返回中没有 id_token")
                 return None
 
-            # 解码 id_token (JWT) payload，不验证签名（HTTPS 已保证传输安全）
-            # 生产环境建议验证 Casdoor 证书
             cert = _load_casdoor_certificate()
-            try:
+            if not cert:
-                payload = jwt.decode(
+                logger.error("Casdoor token 验证证书未配置")
-                    id_token,
+                return None
-                    key=cert or None,
+
-                    options={"verify_signature": bool(cert)},
+            payload = jwt.decode(
-                    audience=settings.CASDOOR_CLIENT_ID,
+                id_token,
-                )
+                key=cert,
-            except JWTError:
+                audience=settings.CASDOOR_CLIENT_ID,
-                # 不验证签名的方式解码
+            )
-                payload = jwt.get_unverified_claims(id_token)
 
             return payload
 

backend/app/core/deps.py

@@ -6,7 +6,7 @@ from app.config import settings
 
 # 处理 sqlite 协议兼容
 db_url = settings.DATABASE_URL
-if db_url.startswith("sqlite"):
+if db_url.startswith("sqlite://"):
     db_url = db_url.replace("sqlite://", "sqlite+aiosqlite://")
 
 engine = create_async_engine(db_url, echo=False, pool_pre_ping=True)
@@ -22,7 +22,14 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]:
 
 
 async def init_db():
-    """创建所有表"""
+    """Apply non-destructive schema upgrades before creating missing tables."""
-    from app.models.device import Base
+    from importlib import import_module
+
+    from app.models import Base
+
+    migration = import_module("migrations.versions.20260803_reliability_monitoring")
+    run_reliability_migration = migration.run_reliability_migration
+
     async with engine.begin() as conn:
+        await conn.run_sync(run_reliability_migration)
         await conn.run_sync(Base.metadata.create_all)

backend/app/models/__init__.py

@@ -1,11 +1,14 @@
-from .device import Device, DeviceTypeEnum
+from .device import Base, Device, DeviceMonitoringPolicy, DeviceTypeEnum
 from .ping_record import PingRecord
 from .alert_event import AlertEvent, AlertTypeEnum
+from .notification_outbox import NotificationOutbox, NotificationStatus
 from .user import User, UserRoleEnum
 
 __all__ = [
-    "Device", "DeviceTypeEnum",
+    "Base",
+    "Device", "DeviceMonitoringPolicy", "DeviceTypeEnum",
     "PingRecord",
     "AlertEvent", "AlertTypeEnum",
+    "NotificationOutbox", "NotificationStatus",
     "User", "UserRoleEnum",
 ]

backend/app/models/alert_event.py

@@ -1,11 +1,15 @@
 import enum
 from datetime import datetime
-from sqlalchemy import Column, Integer, String, Boolean, DateTime, BigInteger, Enum
+from sqlalchemy import BigInteger, Boolean, Column, DateTime, Enum, Integer, String
 from .device import Base
 
 
+PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite")
+
+
 class AlertTypeEnum(str, enum.Enum):
     offline = "offline"           # 设备离线
+    degraded = "degraded"         # 设备丢包故障
     recovered = "recovered"       # 设备恢复
     system = "system"             # 系统告警（如上游断网检测）
 
@@ -14,7 +18,7 @@ class AlertEvent(Base):
     """告警事件表"""
     __tablename__ = "alert_events"
 
-    id = Column(BigInteger, primary_key=True, autoincrement=True)
+    id = Column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
     device_id = Column(Integer, nullable=False, index=True, comment="关联设备 ID")
     alert_type = Column(Enum(AlertTypeEnum), nullable=False, comment="告警类型")
     message = Column(String(1024), default="", comment="告警消息摘要")
@@ -26,6 +30,19 @@ class AlertEvent(Base):
 
     is_resolved = Column(Boolean, default=False, comment="是否已恢复")
     notification_sent = Column(Boolean, default=False, comment="是否已发送通知")
+    notification_attempts = Column(Integer, default=0, comment="通知尝试次数")
+    last_notification_error = Column(
+        String(512),
+        nullable=True,
+        comment="最近一次通知失败原因",
+    )
+    related_event_id = Column(
+        BigInteger,
+        nullable=True,
+        comment="关联的原始或升级告警事件 ID",
+    )
+    previous_status = Column(String(16), nullable=True, comment="状态变更前状态")
+    current_status = Column(String(16), nullable=True, comment="状态变更后状态")
     acknowledged_at = Column(DateTime, nullable=True, comment="用户确认时间")
 
     created_at = Column(DateTime, default=datetime.now, comment="创建时间")

backend/app/models/device.py

@@ -1,8 +1,14 @@
 import enum
+from dataclasses import dataclass
 from datetime import datetime
-from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime, Enum, Text
+from typing import TYPE_CHECKING
+
+from sqlalchemy import Boolean, Column, DateTime, Enum, Float, Integer, String
 from sqlalchemy.orm import DeclarativeBase
 
+if TYPE_CHECKING:
+    from app.config import Settings
+
 
 class Base(DeclarativeBase):
     pass
@@ -16,6 +22,88 @@ class DeviceTypeEnum(str, enum.Enum):
     other = "other"           # 其他
 
 
+@dataclass(frozen=True)
+class DeviceMonitoringPolicy:
+    """Resolved monitoring thresholds for one device.
+
+    Nullable device fields intentionally inherit the runtime defaults so a
+    configuration change can apply to devices that do not need exceptions.
+    """
+
+    probe_packets_per_round: int
+    offline_consecutive_rounds: int
+    degraded_window_rounds: int
+    degraded_loss_percent: float
+    recovery_consecutive_clean_rounds: int
+
+    @staticmethod
+    def _resolve_override(
+        override: object,
+        default: int | float,
+        minimum: int | float,
+        maximum: int | float,
+    ) -> int | float:
+        """Use a bounded persisted override or retain the validated default."""
+        if (
+            isinstance(override, bool)
+            or not isinstance(override, (int, float))
+            or override < minimum
+            or override > maximum
+        ):
+            return default
+        return override
+
+    @classmethod
+    def from_device(
+        cls,
+        device: "Device",
+        settings: "Settings",
+    ) -> "DeviceMonitoringPolicy":
+        """Resolve device overrides without changing the stored device."""
+        return cls(
+            probe_packets_per_round=(
+                cls._resolve_override(
+                    device.probe_packets_per_round,
+                    settings.probe_packets_per_round,
+                    1,
+                    10,
+                )
+            ),
+            offline_consecutive_rounds=(
+                cls._resolve_override(
+                    device.offline_consecutive_rounds,
+                    settings.offline_consecutive_rounds,
+                    1,
+                    10,
+                )
+            ),
+            degraded_window_rounds=(
+                cls._resolve_override(
+                    device.degraded_window_rounds,
+                    settings.degraded_window_rounds,
+                    2,
+                    60,
+                )
+            ),
+            degraded_loss_percent=(
+                cls._resolve_override(
+                    device.degraded_loss_percent,
+                    settings.degraded_loss_percent,
+                    1,
+                    100,
+                )
+            ),
+            recovery_consecutive_clean_rounds=(
+                cls._resolve_override(
+                    device.recovery_consecutive_clean_rounds,
+                    settings.recovery_consecutive_clean_rounds,
+                    1,
+                    20,
+                )
+            ),
+        )
+
+
 class Device(Base):
     __tablename__ = "devices"
 
@@ -32,8 +120,39 @@ class Device(Base):
     alert_threshold = Column(Integer, default=5, comment="连续失败次数判离线")
     is_enabled = Column(Boolean, default=True, comment="是否启用监控")
 
+    # 设备级监测策略覆盖；NULL 时使用运行环境中的全局默认值。
+    probe_packets_per_round = Column(
+        Integer,
+        nullable=True,
+        comment="单轮探测发包数覆盖",
+    )
+    offline_consecutive_rounds = Column(
+        Integer,
+        nullable=True,
+        comment="连续全丢包离线轮数覆盖",
+    )
+    degraded_window_rounds = Column(
+        Integer,
+        nullable=True,
+        comment="故障丢包滑动窗口轮数覆盖",
+    )
+    degraded_loss_percent = Column(
+        Float,
+        nullable=True,
+        comment="故障丢包率阈值覆盖",
+    )
+    recovery_consecutive_clean_rounds = Column(
+        Integer,
+        nullable=True,
+        comment="连续零丢包恢复轮数覆盖",
+    )
+
     # 运行状态
-    current_status = Column(String(16), default="unknown", comment="当前状态: online/offline/unknown")
+    current_status = Column(
+        String(16),
+        default="unknown",
+        comment="当前状态: online/degraded/offline/unknown",
+    )
     consecutive_failures = Column(Integer, default=0, comment="当前连续失败次数")
     last_ping_time = Column(DateTime, nullable=True, comment="最后一次 ping 时间")
     last_online_time = Column(DateTime, nullable=True, comment="最后一次在线时间")

backend/app/models/notification_outbox.py

@@ -0,0 +1,55 @@
+"""Durable delivery queue for alert notifications."""
+
+import enum
+from datetime import datetime
+
+from sqlalchemy import BigInteger, Column, DateTime, Enum, Index, Integer, String, Text
+
+from .device import Base
+
+
+PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite")
+
+
+class NotificationStatus(str, enum.Enum):
+    """Lifecycle states for a notification delivery attempt."""
+
+    pending = "pending"
+    sending = "sending"
+    sent = "sent"
+    failed = "failed"
+
+
+class NotificationOutbox(Base):
+    """A notification retained until a dispatcher records its final outcome."""
+
+    __tablename__ = "notification_outbox"
+
+    id = Column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
+    alert_event_id = Column(BigInteger, nullable=False, index=True, comment="关联告警事件 ID")
+    message_content = Column(Text, nullable=False, comment="待发送的脱敏消息摘要")
+    delivery_scope = Column(String(256), nullable=True, comment="受控投递范围摘要")
+    status = Column(
+        Enum(NotificationStatus),
+        nullable=False,
+        default=NotificationStatus.pending,
+        server_default=NotificationStatus.pending.value,
+        comment="投递状态",
+    )
+    attempt_count = Column(Integer, nullable=False, default=0, comment="投递尝试次数")
+    next_attempt_at = Column(DateTime, nullable=True, comment="下次允许投递时间")
+    locked_at = Column(DateTime, nullable=True, comment="投递器领取时间")
+    sent_at = Column(DateTime, nullable=True, comment="成功投递时间")
+    last_error = Column(String(512), nullable=True, comment="最近一次投递错误摘要")
+    created_at = Column(DateTime, default=datetime.now, nullable=False, comment="创建时间")
+    updated_at = Column(
+        DateTime,
+        default=datetime.now,
+        onupdate=datetime.now,
+        nullable=False,
+        comment="更新时间",
+    )
+
+    __table_args__ = (
+        Index("idx_notification_outbox_status_next_attempt", "status", "next_attempt_at"),
+    )

backend/app/models/ping_record.py

@@ -1,18 +1,32 @@
 from datetime import datetime
-from sqlalchemy import Column, Integer, Float, Boolean, DateTime, BigInteger
+from sqlalchemy import BigInteger, Boolean, Column, DateTime, Float, Index, Integer, String
 from .device import Base
 
 
+PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite")
+
+
 class PingRecord(Base):
     """单次 ping 结果记录"""
     __tablename__ = "ping_records"
 
-    id = Column(BigInteger, primary_key=True, autoincrement=True)
+    id = Column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
     device_id = Column(Integer, nullable=False, index=True)
     is_alive = Column(Boolean, nullable=False, comment="是否通")
     response_time_ms = Column(Float, nullable=True, comment="响应时间毫秒，不通则为 NULL")
     round_num = Column(Integer, nullable=False, comment="轮次编号（从 1 递增）")
     created_at = Column(DateTime, default=datetime.now, index=True, comment="记录时间")
 
+    sent_count = Column(Integer, nullable=True, comment="本轮发包数")
+    received_count = Column(Integer, nullable=True, comment="本轮收包数")
+    packet_loss_percent = Column(Float, nullable=True, comment="本轮丢包率百分比")
+    average_rtt_ms = Column(Float, nullable=True, comment="本轮平均时延毫秒")
+    is_valid = Column(Boolean, nullable=False, default=True, comment="探测结果是否有效")
+    failure_reason = Column(String(512), nullable=True, comment="无效探测的原因")
+
+    __table_args__ = (
+        Index("idx_ping_records_device_created_at", "device_id", "created_at"),
+    )
+
     def __repr__(self):
         return f"<PingRecord(device={self.device_id}, alive={self.is_alive}, rtt={self.response_time_ms})>"

backend/app/services/alerter.py

@@ -1,319 +1,136 @@
-"""
+"""Create durable alert events and their notification outbox entries."""
-告警服务
 
-职责：
-1. 接收 Pinger 的状态变化事件
-2. 判断是否需要发送告警
-3. 上游心跳检测 + 全量离线抑制
-4. 企业微信消息推送（聚合告警）
-5. 记录告警事件到数据库
-"""
-
-import asyncio
-import logging
 from datetime import datetime
-from typing import Optional
-from collections import defaultdict
 
-import httpx
+from sqlalchemy import select
-from sqlalchemy import select, func
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.config import settings
+from app.config import Settings, settings
-from app.models.device import Device
+from app.models import (
-from app.models.alert_event import AlertEvent, AlertTypeEnum
+    AlertEvent,
+    AlertTypeEnum,
+    NotificationOutbox,
+)
 from app.services.pinger import DeviceStateChange
 
-logger = logging.getLogger("pingwatch.alerter")
-
-
-class PendingOfflineAlert:
-    """等待发送的离线告警（用于聚合）"""
-    def __init__(self, device: Device, alert_time: datetime):
-        self.device = device
-        self.alert_time = alert_time
-
 
 class Alerter:
-    """
+    """Translate a persisted device transition into one atomic outbox event."""
-    告警处理器。
 
-    核心逻辑：
+    def __init__(self, runtime_settings: Settings | None = None):
-    - 设备 offline → 收集到待发送队列
+        self._settings = runtime_settings or settings
-    - 每轮结束后检查：
-        a) 上游心跳是否正常？
-        b) 离线率是否 < 90%？
-        c) 满足条件 → 聚合所有待发告警 → 一条企业微信消息
-        d) 不满足 → 丢弃本轮告警，记录系统日志
-    - 设备 recovered → 单独发送恢复通知
-    """
 
-    def __init__(self):
+    async def record_transition(
-        self._pending_alerts: list[PendingOfflineAlert] = []
+        self,
-        self._lock = asyncio.Lock()
+        change: DeviceStateChange,
-        # 上游心跳状态
+        db: AsyncSession,
-        self._upstream_failures = 0
+    ) -> AlertEvent:
-        self._upstream_available = True
+        """Stage an event and notification without committing the caller session."""
+        occurred_at = change.device.last_ping_time or datetime.now()
+        event_type = AlertTypeEnum(change.event_type or change.new_status)
+        prior_event = await self._find_prior_open_event(change, db)
+        duration_minutes = self._close_prior_event(prior_event, occurred_at)
+        event = AlertEvent(
+            device_id=change.device.id,
+            alert_type=event_type,
+            message=change.reason,
+            start_at=occurred_at,
+            is_resolved=event_type == AlertTypeEnum.recovered,
+            duration_minutes=(
+                duration_minutes
+                if event_type == AlertTypeEnum.recovered
+                else None
+            ),
+            related_event_id=(prior_event.id if prior_event is not None else None),
+            previous_status=change.old_status,
+            current_status=change.new_status,
+        )
+        db.add(event)
+        await db.flush()
 
-    async def on_state_change(self, change: DeviceStateChange, db: AsyncSession):
+        if prior_event is not None:
-        """Pinger 状态变化回调"""
+            prior_event.related_event_id = event.id
-        if change.new_status == "offline":
-            async with self._lock:
-                self._pending_alerts.append(
-                    PendingOfflineAlert(device=change.device, alert_time=datetime.now())
-                )
 
-        elif change.new_status == "online" and change.old_status == "offline":
+        db.add(
-            # 设备恢复，立即记录并发送恢复通知
+            NotificationOutbox(
-            await self._handle_recovery(change.device, db)
+                alert_event_id=event.id,
-
+                message_content=self._render_notification(
-    async def flush_pending(self, db: AsyncSession, total_device_count: int):
+                    change,
-        """
+                    occurred_at,
-        每轮结束时调用：处理待发送的离线告警。
+                    duration_minutes,
-        判断是否应该抑制告警，然后发送或丢弃。
+                ),
-        """
+                delivery_scope=(
-        async with self._lock:
+                    self._settings.WECOM_TO_PARTY.strip() or "@all"
-            if not self._pending_alerts:
+                ),
-                return
-            pending = self._pending_alerts.copy()
-            self._pending_alerts.clear()
-
-        # 1. 检查上游心跳
-        upstream_ok = await self._check_upstream()
-
-        # 2. 计算离线率
-        offline_count = len(pending)
-        offline_ratio = offline_count / max(total_device_count, 1)
-
-        # 3. 抑制条件
-        suppressed = False
-        suppress_reason = ""
-
-        if not upstream_ok:
-            suppressed = True
-            suppress_reason = "上游网络不可达（监控节点可能断网）"
-        elif offline_ratio >= settings.OFFLINE_SUPPRESS_RATIO:
-            suppressed = True
-            suppress_reason = f"离线率 {offline_ratio:.0%} >= {settings.OFFLINE_SUPPRESS_RATIO:.0%}，疑似监控节点断网"
-
-        if suppressed:
-            logger.warning(
-                f"告警抑制: {suppress_reason}，"
-                f"本轮 {offline_count} 条告警已丢弃"
             )
-            # 记录系统告警
+        )
-            db.add(AlertEvent(
+        await db.flush()
-                device_id=0,
+        return event
-                alert_type=AlertTypeEnum.system,
-                message=f"告警抑制: {suppress_reason}，丢弃 {offline_count} 条离线告警",
-                start_at=datetime.now(),
-                is_resolved=True,
-                notification_sent=False,
-            ))
-            await db.commit()
-            return
 
-        # 4. 发送聚合告警
+    @staticmethod
-        if pending:
+    async def _find_prior_open_event(
-            await self._send_aggregated_alert(pending, db)
+        change: DeviceStateChange,
-
+        db: AsyncSession,
-    async def _check_upstream(self) -> bool:
+    ) -> AlertEvent | None:
-        """
+        """Find only the incident that this escalation or recovery supersedes."""
-        上游心跳检测。
+        event_type = change.event_type or change.new_status
-        连续 UPSTREAM_PING_THRESHOLD 次失败才判定为上游断网。
+        if event_type == AlertTypeEnum.offline.value and change.old_status == "degraded":
-        """
+            alert_types = (AlertTypeEnum.degraded,)
-        try:
+        elif event_type == AlertTypeEnum.recovered.value:
-            import subprocess
+            alert_types = (AlertTypeEnum.offline, AlertTypeEnum.degraded)
-            proc = await asyncio.create_subprocess_exec(
-                "ping", "-c", "1", "-W", "3",
-                settings.UPSTREAM_PING_TARGET,
-                stdout=subprocess.DEVNULL,
-                stderr=subprocess.DEVNULL,
-            )
-            await proc.wait()
-            if proc.returncode == 0:
-                self._upstream_failures = 0
-                self._upstream_available = True
-                return True
-            else:
-                self._upstream_failures += 1
-                if self._upstream_failures >= settings.UPSTREAM_PING_THRESHOLD:
-                    self._upstream_available = False
-                    return False
-                # 未达到阈值，认为上游还可用
-                return True
-        except Exception as e:
-            logger.error(f"上游心跳检测异常: {e}")
-            return True  # 异常时保守地允许告警
-
-    async def _send_aggregated_alert(self, alerts: list[PendingOfflineAlert], db: AsyncSession):
-        """发送聚合离线告警"""
-        now = datetime.now()
-
-        # 构建企业微信消息
-        if len(alerts) == 1:
-            a = alerts[0]
-            msg = self._build_offline_message_single(a.device, a.alert_time)
         else:
-            msg = self._build_offline_message_batch(alerts)
+            return None
 
-        # 发送
+        return await db.scalar(
-        success = await self._send_wecom_message(msg)
-
-        # 记录告警事件
-        for a in alerts:
-            db.add(AlertEvent(
-                device_id=a.device.id,
-                alert_type=AlertTypeEnum.offline,
-                message=a.device.name,
-                start_at=a.alert_time,
-                is_resolved=False,
-                notification_sent=success,
-            ))
-
-        await db.commit()
-
-        if success:
-            logger.info(f"已推送离线告警: {len(alerts)} 台设备")
-        else:
-            logger.error(f"企业微信推送失败: {len(alerts)} 台设备")
-
-    async def _handle_recovery(self, device: Device, db: AsyncSession):
-        """处理设备恢复"""
-        now = datetime.now()
-
-        # 查找未解决的离线事件
-        result = await db.execute(
             select(AlertEvent)
-            .where(AlertEvent.device_id == device.id)
+            .where(AlertEvent.device_id == change.device.id)
-            .where(AlertEvent.alert_type == AlertTypeEnum.offline)
+            .where(AlertEvent.alert_type.in_(alert_types))
-            .where(AlertEvent.is_resolved == False)
+            .where(AlertEvent.is_resolved.is_(False))
-            .order_by(AlertEvent.created_at.desc())
+            .order_by(AlertEvent.start_at.desc(), AlertEvent.id.desc())
             .limit(1)
         )
-        event = result.scalar_one_or_none()
 
-        duration_minutes = None
+    @staticmethod
-        if event:
+    def _close_prior_event(
-            delta = now - event.start_at
+        event: AlertEvent | None,
-            duration_minutes = int(delta.total_seconds() / 60)
+        occurred_at: datetime,
-            event.end_at = now
+    ) -> int | None:
-            event.duration_minutes = duration_minutes
+        """Resolve one incident and return a non-negative minute duration."""
-            event.is_resolved = True
+        if event is None:
-
+            return None
-        # 发送恢复通知
+        duration_minutes = max(
-        msg = self._build_recovery_message(device, now, duration_minutes)
+            0,
-        success = await self._send_wecom_message(msg)
+            int((occurred_at - event.start_at).total_seconds() / 60),
-
-        if event:
-            event.notification_sent = success
-
-        await db.commit()
-
-        if success:
-            logger.info(f"已推送恢复通知: {device.name}")
-        else:
-            logger.error(f"恢复通知推送失败: {device.name}")
-
-    # ---------- 消息格式化 ----------
-
-    def _build_offline_message_single(self, device: Device, alert_time: datetime) -> str:
-        """单台设备离线消息"""
-        time_str = alert_time.strftime("%Y-%m-%d %H:%M:%S")
-        return (
-            f"⛔ 设备离线啦！\n"
-            f"地址：{device.location or '未知'}\n"
-            f"时间：{time_str}\n"
-            f"项目：{device.project_name or '未分组'}\n"
-            f"设备类型：{self._fmt_device_type(device.device_type)}\n"
-            f"IP地址：{device.ip}"
         )
+        event.end_at = occurred_at
+        event.duration_minutes = duration_minutes
+        event.is_resolved = True
+        return duration_minutes
 
-    def _build_offline_message_batch(self, alerts: list[PendingOfflineAlert]) -> str:
+    @staticmethod
-        """多台设备聚合离线消息"""
+    def _render_notification(
-        now = alerts[0].alert_time
+        change: DeviceStateChange,
-        time_str = now.strftime("%Y-%m-%d %H:%M:%S")
+        occurred_at: datetime,
-        lines = [f"⛔ 设备离线啦！（共 {len(alerts)} 台）\n"]
+        duration_minutes: int | None = None,
-
+    ) -> str:
-        for a in alerts:
+        """Render the bounded device and transition details needed by operators."""
-            dev = a.device
+        labels = {
-            lines.append(
+            "offline": "设备离线",
-                f"地址：{dev.location or '未知'}\n"
+            "degraded": "链路质量故障",
-                f"时间：{a.alert_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
+            "recovered": "设备恢复",
-                f"项目：{dev.project_name or '未分组'}\n"
+            "system": "监控系统异常",
-                f"设备类型：{self._fmt_device_type(dev.device_type)}\n"
-                f"IP地址：{dev.ip}\n"
-                f"{'---' if len(alerts) > 1 else ''}"
-            )
-
-        return "\n".join(lines).rstrip("---\n")
-
-    def _build_recovery_message(self, device: Device, recover_time: datetime, duration: Optional[int]) -> str:
-        """设备恢复消息"""
-        time_str = recover_time.strftime("%Y-%m-%d %H:%M:%S")
-        duration_str = f"{duration}分钟" if duration is not None else "未知"
-        return (
-            f"✅ 设备恢复在线！\n"
-            f"地址：{device.location or '未知'}\n"
-            f"时间：{time_str}\n"
-            f"项目：{device.project_name or '未分组'}\n"
-            f"设备类型：{self._fmt_device_type(device.device_type)}\n"
-            f"IP地址：{device.ip}\n"
-            f"离线时长：{duration_str}"
-        )
-
-    def _fmt_device_type(self, dtype) -> str:
-        mapping = {
-            "server": "服务器",
-            "olt": "OLT",
-            "switch": "交换机",
-            "firewall": "防火墙",
-            "other": "其他设备",
         }
-        return mapping.get(str(dtype), str(dtype))
+        event_type = change.event_type or change.new_status
-
+        device = change.device
-    # ---------- 企业微信推送 ----------
+        lines = [
-
+            f"【{labels.get(event_type, event_type)}】",
-    async def _get_wecom_token(self) -> Optional[str]:
+            f"设备：{device.name}",
-        """获取企业微信 access_token"""
+            f"IP：{device.ip}",
-        url = (
+            f"位置：{device.location or '未配置'}",
-            f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
+            f"项目：{device.project_name or '未分组'}",
-            f"?corpid={settings.WECOM_CORP_ID}"
+            f"丢包摘要：{change.reason or '无'}",
-            f"&corpsecret={settings.WECOM_APP_SECRET}"
+            f"时间：{occurred_at:%Y-%m-%d %H:%M:%S}",
-        )
+        ]
-        async with httpx.AsyncClient(timeout=10) as client:
+        if duration_minutes is not None:
-            resp = await client.get(url)
+            lines.append(f"持续时间：{duration_minutes} 分钟")
-            data = resp.json()
+        return "\n".join(lines)
-            if data.get("errcode") == 0:
-                return data["access_token"]
-            else:
-                logger.error(f"获取企业微信 token 失败: {data}")
-                return None
-
-    async def _send_wecom_message(self, content: str) -> bool:
-        """发送企业微信应用消息"""
-        if not settings.WECOM_CORP_ID or not settings.WECOM_APP_SECRET:
-            logger.warning("企业微信未配置，跳过推送")
-            logger.info(f"[模拟推送] {content}")
-            return True  # 开发模式
-
-        token = await self._get_wecom_token()
-        if not token:
-            return False
-
-        url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
-        payload = {
-            "touser": "@all",
-            "msgtype": "text",
-            "agentid": settings.WECOM_AGENT_ID,
-            "text": {"content": content},
-            "safe": 0,
-        }
-
-        async with httpx.AsyncClient(timeout=10) as client:
-            resp = await client.post(url, json=payload)
-            data = resp.json()
-            if data.get("errcode") != 0:
-                logger.error(f"发送企业微信消息失败: {data}")
-                return False
-            return True

backend/app/services/fping_runner.py

@@ -0,0 +1,178 @@
+"""Execute and parse multi-packet fping probes without shell interpolation."""
+
+import asyncio
+from dataclasses import dataclass
+import re
+from typing import Iterable
+
+
+FPING_RESULT_PATTERN = re.compile(r"^\s*(\S+)\s+:\s*(.*?)\s*$")
+
+
+@dataclass(frozen=True)
+class ProbeResult:
+    """Aggregate result for one IP in one monitoring round."""
+
+    ip: str
+    sent_count: int
+    received_count: int
+    average_rtt_ms: float | None
+    is_valid: bool
+    failure_reason: str | None = None
+
+    @property
+    def packet_loss_percent(self) -> float:
+        """Return packet loss as a percentage of the attempted probes."""
+        if self.sent_count <= 0:
+            return 0.0
+        return (
+            (self.sent_count - self.received_count)
+            / self.sent_count
+            * 100.0
+        )
+
+
+def _invalid_result(
+    ip: str,
+    packets_per_round: int,
+    reason: str,
+) -> ProbeResult:
+    """Build an invalid aggregate that cannot resemble measured packet loss."""
+    return ProbeResult(
+        ip=ip,
+        sent_count=packets_per_round,
+        received_count=0,
+        average_rtt_ms=None,
+        is_valid=False,
+        failure_reason=reason,
+    )
+
+
+def parse_fping_count_output(
+    output: str,
+    expected_ips: set[str],
+    packets_per_round: int,
+) -> dict[str, ProbeResult]:
+    """Parse ``fping -C`` reply tokens for every expected IP.
+
+    A numeric token is a measured RTT and ``-`` is packet loss. Missing,
+    duplicated, or malformed lines are represented as invalid aggregates.
+    """
+    if packets_per_round <= 0:
+        raise ValueError("packets_per_round must be positive")
+
+    parsed_lines: dict[str, list[str]] = {}
+    duplicate_ips: set[str] = set()
+    for line in output.splitlines():
+        match = FPING_RESULT_PATTERN.match(line)
+        if match is None:
+            continue
+        ip, replies = match.groups()
+        if ip not in expected_ips:
+            continue
+        if ip in parsed_lines:
+            duplicate_ips.add(ip)
+            continue
+        parsed_lines[ip] = replies.split()
+
+    results: dict[str, ProbeResult] = {}
+    for ip in expected_ips:
+        tokens = parsed_lines.get(ip)
+        if tokens is None:
+            results[ip] = _invalid_result(
+                ip,
+                packets_per_round,
+                "missing fping output",
+            )
+            continue
+        if ip in duplicate_ips or len(tokens) != packets_per_round:
+            results[ip] = _invalid_result(
+                ip,
+                packets_per_round,
+                "malformed fping replies",
+            )
+            continue
+
+        reply_times: list[float] = []
+        malformed = False
+        for token in tokens:
+            if token == "-":
+                continue
+            try:
+                reply_time = float(token)
+            except ValueError:
+                malformed = True
+                break
+            if reply_time < 0:
+                malformed = True
+                break
+            reply_times.append(reply_time)
+
+        if malformed:
+            results[ip] = _invalid_result(
+                ip,
+                packets_per_round,
+                "malformed fping replies",
+            )
+            continue
+
+        average_rtt_ms = (
+            sum(reply_times) / len(reply_times)
+            if reply_times
+            else None
+        )
+        results[ip] = ProbeResult(
+            ip=ip,
+            sent_count=packets_per_round,
+            received_count=len(reply_times),
+            average_rtt_ms=average_rtt_ms,
+            is_valid=True,
+        )
+
+    return results
+
+
+async def run_fping_count(
+    ips: Iterable[str],
+    packets_per_round: int,
+    timeout_ms: int,
+    executable: str,
+) -> dict[str, ProbeResult]:
+    """Run one fping count probe and return an aggregate for every IP."""
+    ip_list = list(ips)
+    if not ip_list:
+        return {}
+
+    command = [
+        executable,
+        "-C",
+        str(packets_per_round),
+        "-q",
+        "-t",
+        str(timeout_ms),
+        *ip_list,
+    ]
+    try:
+        process = await asyncio.create_subprocess_exec(
+            *command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+    except Exception as error:
+        reason = f"fping execution failed: {type(error).__name__}"
+        return {
+            ip: _invalid_result(ip, packets_per_round, reason)
+            for ip in ip_list
+        }
+
+    output = "\n".join(
+        stream.decode("utf-8", errors="replace")
+        for stream in (stdout, stderr)
+        if stream
+    )
+    return parse_fping_count_output(
+        output,
+        set(ip_list),
+        packets_per_round,
+    )

backend/app/services/notification_dispatcher.py

@@ -0,0 +1,322 @@
+"""Deliver pending notifications from the durable outbox."""
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import Callable, Protocol
+
+import httpx
+from sqlalchemy import or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import Settings, settings
+from app.models import AlertEvent, NotificationOutbox, NotificationStatus
+
+
+logger = logging.getLogger("pingwatch.notification_dispatcher")
+
+# A small count caps one WeCom text request while still aggregating alert bursts.
+MAX_EVENTS_PER_MESSAGE = 5
+_MESSAGE_SEPARATOR = "\n\n---\n\n"
+_TOKEN_EXPIRY_MARGIN_SECONDS = 60
+_HTTP_TIMEOUT = httpx.Timeout(connect=5.0, read=10.0, write=10.0, pool=5.0)
+_TOKEN_URL = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
+_MESSAGE_URL = "https://qyapi.weixin.qq.com/cgi-bin/message/send"
+
+
+@dataclass(frozen=True)
+class DeliveryResult:
+    """Sanitized delivery outcome returned by a notification transport."""
+
+    success: bool
+    error: str | None = None
+    retryable: bool = False
+
+
+@dataclass(frozen=True)
+class DispatchSummary:
+    """Counts produced by one bounded dispatcher pass."""
+
+    sent: int = 0
+    rescheduled: int = 0
+    failed: int = 0
+
+
+class TextNotificationClient(Protocol):
+    """Minimal transport boundary used by the persistent dispatcher."""
+
+    async def send_text(self, content: str) -> DeliveryResult:
+        """Deliver one text message without exposing credentials."""
+
+
+@dataclass(frozen=True)
+class _TokenResult:
+    token: str | None
+    error: str | None = None
+    retryable: bool = False
+
+
+class WeComClient:
+    """Minimal WeCom application client with an in-process token cache."""
+
+    def __init__(
+        self,
+        runtime_settings: Settings | None = None,
+        *,
+        http_client: httpx.AsyncClient | None = None,
+        now: Callable[[], datetime] | None = None,
+    ):
+        self._settings = runtime_settings or settings
+        self._http = http_client or httpx.AsyncClient(timeout=_HTTP_TIMEOUT)
+        self._owns_http_client = http_client is None
+        self._now = now or datetime.now
+        self._access_token: str | None = None
+        self._access_token_expires_at: datetime | None = None
+        self._token_lock = asyncio.Lock()
+
+    async def aclose(self) -> None:
+        """Close only the HTTP client created by this service instance."""
+        if self._owns_http_client:
+            await self._http.aclose()
+
+    async def send_text(self, content: str) -> DeliveryResult:
+        """Deliver text while returning only sanitized failure evidence."""
+        if not self._settings.wecom_notification_enabled:
+            return DeliveryResult(False, "delivery disabled", False)
+
+        token_result = await self._get_access_token()
+        if token_result.token is None:
+            return DeliveryResult(
+                False,
+                token_result.error,
+                token_result.retryable,
+            )
+
+        payload = {
+            "msgtype": "text",
+            "agentid": self._settings.WECOM_AGENT_ID,
+            "text": {"content": content},
+            "safe": 0,
+        }
+        to_party = self._settings.WECOM_TO_PARTY.strip()
+        if to_party:
+            payload["toparty"] = to_party
+        else:
+            payload["touser"] = "@all"
+
+        try:
+            response = await self._http.post(
+                _MESSAGE_URL,
+                params={"access_token": token_result.token},
+                json=payload,
+            )
+        except (httpx.TimeoutException, httpx.TransportError):
+            logger.warning("WeCom message transport failure")
+            return DeliveryResult(False, "transport error", True)
+
+        if response.status_code == 429 or response.status_code >= 500:
+            logger.warning(
+                "WeCom message transient HTTP failure status=%s",
+                response.status_code,
+            )
+            return DeliveryResult(False, f"HTTP {response.status_code}", True)
+        if response.status_code >= 400:
+            logger.warning(
+                "WeCom message HTTP failure status=%s",
+                response.status_code,
+            )
+            return DeliveryResult(False, f"HTTP {response.status_code}", False)
+
+        try:
+            data = response.json()
+            if not isinstance(data, dict):
+                raise TypeError("response JSON must be an object")
+            error_code = int(data.get("errcode", -1))
+        except (TypeError, ValueError):
+            logger.warning("WeCom message returned invalid JSON")
+            return DeliveryResult(False, "invalid response", True)
+
+        if error_code == 0:
+            return DeliveryResult(True)
+
+        if error_code in {40014, 42001}:
+            self._access_token = None
+            self._access_token_expires_at = None
+        retryable = error_code in {-1, 40014, 42001, 45009}
+        logger.warning("WeCom message rejected errcode=%s", error_code)
+        return DeliveryResult(
+            False,
+            f"WeCom errcode {error_code}",
+            retryable,
+        )
+
+    async def _get_access_token(self) -> _TokenResult:
+        """Return a cached token or refresh it once under a process-local lock."""
+        if self._token_is_valid():
+            return _TokenResult(self._access_token)
+
+        async with self._token_lock:
+            if self._token_is_valid():
+                return _TokenResult(self._access_token)
+            try:
+                response = await self._http.get(
+                    _TOKEN_URL,
+                    params={
+                        "corpid": self._settings.WECOM_CORP_ID,
+                        "corpsecret": self._settings.WECOM_APP_SECRET,
+                    },
+                )
+            except (httpx.TimeoutException, httpx.TransportError):
+                logger.warning("WeCom token transport failure")
+                return _TokenResult(None, "token transport error", True)
+
+            if response.status_code == 429 or response.status_code >= 500:
+                logger.warning(
+                    "WeCom token transient HTTP failure status=%s",
+                    response.status_code,
+                )
+                return _TokenResult(None, f"token HTTP {response.status_code}", True)
+            if response.status_code >= 400:
+                logger.warning(
+                    "WeCom token HTTP failure status=%s",
+                    response.status_code,
+                )
+                return _TokenResult(None, f"token HTTP {response.status_code}", False)
+
+            try:
+                data = response.json()
+                if not isinstance(data, dict):
+                    raise TypeError("response JSON must be an object")
+                error_code = int(data.get("errcode", -1))
+                access_token = data.get("access_token")
+                expires_in = int(data.get("expires_in", 0))
+            except (TypeError, ValueError):
+                logger.warning("WeCom token endpoint returned invalid JSON")
+                return _TokenResult(None, "invalid token response", True)
+
+            if error_code != 0 or not access_token or expires_in <= 0:
+                logger.warning("WeCom token request rejected errcode=%s", error_code)
+                return _TokenResult(
+                    None,
+                    f"WeCom token errcode {error_code}",
+                    error_code == -1,
+                )
+
+            self._access_token = str(access_token)
+            cache_seconds = max(0, expires_in - _TOKEN_EXPIRY_MARGIN_SECONDS)
+            self._access_token_expires_at = self._now() + timedelta(
+                seconds=cache_seconds
+            )
+            return _TokenResult(self._access_token)
+
+    def _token_is_valid(self) -> bool:
+        return (
+            self._access_token is not None
+            and self._access_token_expires_at is not None
+            and self._now() < self._access_token_expires_at
+        )
+
+
+class NotificationDispatcher:
+    """Claim due outbox rows and persist their delivery outcome."""
+
+    def __init__(
+        self,
+        client: TextNotificationClient,
+        runtime_settings: Settings | None = None,
+    ):
+        self._client = client
+        self._settings = runtime_settings or settings
+
+    async def dispatch_due(
+        self,
+        db: AsyncSession,
+        now: datetime,
+    ) -> DispatchSummary:
+        """Attempt each currently due message once without committing the session."""
+        rows = list(
+            (
+                await db.execute(
+                    select(NotificationOutbox)
+                    .where(NotificationOutbox.status == NotificationStatus.pending)
+                    .where(
+                        or_(
+                            NotificationOutbox.next_attempt_at.is_(None),
+                            NotificationOutbox.next_attempt_at <= now,
+                        )
+                    )
+                    .order_by(NotificationOutbox.created_at, NotificationOutbox.id)
+                )
+            )
+            .scalars()
+            .all()
+        )
+
+        event_ids = [row.alert_event_id for row in rows]
+        events = {}
+        if event_ids:
+            events = {
+                event.id: event
+                for event in (
+                    (
+                        await db.execute(
+                            select(AlertEvent).where(AlertEvent.id.in_(event_ids))
+                        )
+                    )
+                    .scalars()
+                    .all()
+                )
+            }
+
+        sent = rescheduled = failed = 0
+        for start in range(0, len(rows), MAX_EVENTS_PER_MESSAGE):
+            batch = rows[start : start + MAX_EVENTS_PER_MESSAGE]
+            for row in batch:
+                row.status = NotificationStatus.sending
+                row.locked_at = now
+            try:
+                result = await self._client.send_text(
+                    _MESSAGE_SEPARATOR.join(row.message_content for row in batch)
+                )
+            except Exception as exc:  # Transport implementations must not strand rows.
+                logger.warning(
+                    "Notification client raised type=%s",
+                    type(exc).__name__,
+                )
+                result = DeliveryResult(False, "notification client error", True)
+
+            for row in batch:
+                row.attempt_count += 1
+                event = events.get(row.alert_event_id)
+                if result.success:
+                    row.status = NotificationStatus.sent
+                    row.sent_at = now
+                    row.next_attempt_at = None
+                    row.last_error = None
+                    sent += 1
+                elif (
+                    result.retryable
+                    and row.attempt_count
+                    < self._settings.wecom_notification_max_attempts
+                ):
+                    delay = self._settings.wecom_retry_base_seconds * (
+                        2 ** (row.attempt_count - 1)
+                    )
+                    row.status = NotificationStatus.pending
+                    row.next_attempt_at = now + timedelta(seconds=delay)
+                    row.last_error = (result.error or "delivery failed")[:512]
+                    rescheduled += 1
+                else:
+                    row.status = NotificationStatus.failed
+                    row.next_attempt_at = None
+                    row.last_error = (result.error or "delivery failed")[:512]
+                    failed += 1
+
+                if event is not None:
+                    event.notification_sent = result.success
+                    event.notification_attempts = row.attempt_count
+                    event.last_notification_error = row.last_error
+
+        await db.flush()
+        return DispatchSummary(sent=sent, rescheduled=rescheduled, failed=failed)

backend/app/services/pinger.py

@@ -1,242 +1,252 @@
-"""
+"""Persist batched probe summaries and apply device health transitions."""
-异步 Ping 引擎
 
-核心逻辑：
-1. 每轮从数据库加载所有启用设备，批量 fping
-2. 记录每台设备本轮 ping 结果（存活/响应时间）
-3. 状态机管理设备状态，判断是否从 online→offline 或 offline→online
-4. 结果通过回调或队列通知 alerter
-"""
-
-import asyncio
-import subprocess
-import time
 import logging
-from datetime import datetime
+import time
-from typing import Optional, Callable, Awaitable
 from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime
 
-from sqlalchemy import select, update
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.config import settings
+from app.config import Settings, settings
-from app.models.device import Device, DeviceTypeEnum
+from app.models.device import Device, DeviceMonitoringPolicy
 from app.models.ping_record import PingRecord
+from app.services.fping_runner import ProbeResult, run_fping_count
+from app.services.state_machine import StateDecision, evaluate_health
+
 
 logger = logging.getLogger("pingwatch.pinger")
 
 
-class PingResult:
+@dataclass(frozen=True)
-    """单台设备一轮 ping 的结果"""
-    def __init__(self, device_id: int, is_alive: bool, response_time_ms: Optional[float] = None):
-        self.device_id = device_id
-        self.is_alive = is_alive
-        self.response_time_ms = response_time_ms
-
-
 class DeviceStateChange:
-    """设备状态变化事件"""
+    """A persisted device health transition produced by one probe round."""
-    def __init__(self, device: Device, old_status: str, new_status: str, consecutive_failures: int):
+
-        self.device = device
+    device: Device
-        self.old_status = old_status
+    old_status: str
-        self.new_status = new_status
+    new_status: str
-        self.consecutive_failures = consecutive_failures
+    consecutive_failures: int
+    event_type: str | None = None
+    reason: str = ""
 
 
 class Pinger:
-    """
+    """Run multi-packet fping probes and update monitored device health."""
-    Ping 引擎，使用 fping 批量并发检测。
-    对所有设备进 ping，返回存活状态和响应时间。
-    """
 
-    def __init__(self):
+    def __init__(self, runtime_settings: Settings | None = None):
+        self._settings = runtime_settings or settings
         self._round_num = 0
-        self._on_state_change: Optional[Callable[[DeviceStateChange], Awaitable[None]]] = None
 
-    def on_state_change(self, callback: Callable[[DeviceStateChange], Awaitable[None]]):
+    async def run_one_round(
-        """注册状态变化回调"""
+        self,
-        self._on_state_change = callback
+        db: AsyncSession,
-
+    ) -> list[DeviceStateChange]:
-    async def run_one_round(self, db: AsyncSession) -> list[PingResult]:
+        """Persist one summary per enabled device and commit state changes once."""
-        """
-        执行一轮 ping 检测：
-        1. 加载所有启用设备
-        2. 批量 fping
-        3. 记录结果
-        4. 更新设备状态
-        """
         self._round_num += 1
         round_num = self._round_num
-
+        query_result = await db.execute(
-        # 1. 加载启用设备
+            select(Device).where(Device.is_enabled.is_(True))
-        result = await db.execute(
-            select(Device).where(Device.is_enabled == True)
         )
-        devices = list(result.scalars().all())
+        devices = list(query_result.scalars().all())
-
         if not devices:
-            logger.info(f"[Round {round_num}] 没有启用的设备")
+            logger.info("[Round %s] no enabled devices", round_num)
             return []
 
-        logger.info(f"[Round {round_num}] 开始检测 {len(devices)} 台设备")
+        started_at = time.monotonic()
-
+        policies = {
-        # 2. 批量 ping
+            device.id: DeviceMonitoringPolicy.from_device(
-        start_time = time.time()
+                device,
-        ip_to_device = {d.ip: d for d in devices}
+                self._settings,
-        ip_list = list(ip_to_device.keys())
-
-        ping_results_map = await self._batch_ping(ip_list)
-
-        # 3. 构造结果
-        results: list[PingResult] = []
-        for ip, dev in ip_to_device.items():
-            is_alive, rtt = ping_results_map.get(ip, (False, None))
-            results.append(PingResult(device_id=dev.id, is_alive=is_alive, response_time_ms=rtt))
-
-        elapsed = time.time() - start_time
-        alive_count = sum(1 for r in results if r.is_alive)
-
-        # 4. 批量写入 ping_records
-        now = datetime.now()
-        records = [
-            PingRecord(
-                device_id=r.device_id,
-                is_alive=r.is_alive,
-                response_time_ms=r.response_time_ms,
-                round_num=round_num,
-                created_at=now,
             )
-            for r in results
+            for device in devices
-        ]
+        }
-        db.add_all(records)
+        probe_results = await self._probe_by_packet_count(devices, policies)
+        observed_at = datetime.now()
+
+        records_by_device: dict[int, PingRecord] = {}
+        for device in devices:
+            policy = policies[device.id]
+            probe = probe_results.get(
+                (device.id, policy.probe_packets_per_round)
+            )
+            if probe is None:
+                probe = ProbeResult(
+                    ip=device.ip,
+                    sent_count=policy.probe_packets_per_round,
+                    received_count=0,
+                    average_rtt_ms=None,
+                    is_valid=False,
+                    failure_reason="missing runner result",
+                )
+            record = self._to_record(
+                device=device,
+                probe=probe,
+                round_num=round_num,
+                observed_at=observed_at,
+            )
+            records_by_device[device.id] = record
+            db.add(record)
+
         await db.flush()
 
-        # 5. 更新设备状态（状态机）
+        changes: list[DeviceStateChange] = []
-        device_map = {d.id: d for d in devices}
+        for device in devices:
-        for r in results:
+            record = records_by_device[device.id]
-            dev = device_map.get(r.device_id)
+            device.last_ping_time = observed_at
-            if not dev:
+            if not record.is_valid:
                 continue
 
-            old_status = dev.current_status
+            policy = policies[device.id]
-            if r.is_alive:
+            recent = await self._load_recent_valid(
-                dev.consecutive_failures = 0
+                db,
-                dev.last_ping_time = now
+                device.id,
-                dev.last_online_time = now
+                policy,
-                dev.current_status = "online"
+            )
-            else:
+            decision = evaluate_health(
-                dev.consecutive_failures = (dev.consecutive_failures or 0) + 1
+                device.current_status or "unknown",
-                dev.last_ping_time = now
+                recent,
-                if dev.consecutive_failures >= dev.alert_threshold:
+                policy,
-                    if dev.current_status != "offline":
+            )
-                        dev.current_status = "offline"
+            self._update_observation_fields(device, recent, observed_at)
-                        dev.last_offline_time = now
+            change = self._apply_decision(
-                else:
+                device,
-                    if dev.current_status == "online":
+                decision,
-                        dev.current_status = "checking"
+                observed_at,
+            )
+            if change is not None:
+                changes.append(change)
 
-            # 状态变化回调
+        # The scheduler records transition events and commits the complete
-            if old_status != dev.current_status and self._on_state_change:
+        # probe/state/outbox unit.  A direct caller retains the same ownership.
-                change = DeviceStateChange(
+        await db.flush()
-                    device=dev,
-                    old_status=old_status,
-                    new_status=dev.current_status,
-                    consecutive_failures=dev.consecutive_failures,
-                )
-                await self._on_state_change(change)
-
-        await db.commit()
 
+        elapsed = time.monotonic() - started_at
         logger.info(
-            f"[Round {round_num}] 完成: {alive_count}/{len(devices)} 在线, "
+            "[Round %s] persisted %s device summaries with %s transitions "
-            f"耗时 {elapsed:.2f}s"
+            "in %.2fs",
+            round_num,
+            len(devices),
+            len(changes),
+            elapsed,
         )
+        return changes
+
+    async def _probe_by_packet_count(
+        self,
+        devices: list[Device],
+        policies: dict[int, DeviceMonitoringPolicy],
+    ) -> dict[tuple[int, int], ProbeResult]:
+        """Run one argv-safe batch for each distinct packet-count policy."""
+        groups: dict[int, list[Device]] = defaultdict(list)
+        for device in devices:
+            groups[policies[device.id].probe_packets_per_round].append(device)
+
+        results: dict[tuple[int, int], ProbeResult] = {}
+        timeout_ms = int(self._settings.PING_TIMEOUT_SECONDS * 1000)
+        for packet_count, grouped_devices in groups.items():
+            unique_ips = list(dict.fromkeys(
+                device.ip for device in grouped_devices
+            ))
+            batch = await run_fping_count(
+                unique_ips,
+                packets_per_round=packet_count,
+                timeout_ms=timeout_ms,
+                executable=self._settings.FPING_PATH,
+            )
+            for device in grouped_devices:
+                probe = batch.get(device.ip)
+                if probe is not None:
+                    results[(device.id, packet_count)] = probe
         return results
 
-    async def _batch_ping(self, ip_list: list[str]) -> dict[str, tuple[bool, Optional[float]]]:
+    @staticmethod
-        """
+    def _to_record(
-        使用 fping 批量 ping
+        device: Device,
-        返回: { ip: (is_alive, response_time_ms) }
+        probe: ProbeResult,
-        """
+        round_num: int,
-        if not ip_list:
+        observed_at: datetime,
-            return {}
+    ) -> PingRecord:
+        """Convert a runner aggregate into the persisted compatibility model."""
+        return PingRecord(
+            device_id=device.id,
+            is_alive=probe.is_valid and probe.received_count > 0,
+            response_time_ms=probe.average_rtt_ms,
+            round_num=round_num,
+            created_at=observed_at,
+            sent_count=probe.sent_count,
+            received_count=probe.received_count,
+            packet_loss_percent=(
+                probe.packet_loss_percent if probe.is_valid else None
+            ),
+            average_rtt_ms=probe.average_rtt_ms,
+            is_valid=probe.is_valid,
+            failure_reason=probe.failure_reason,
+        )
 
-        try:
+    @staticmethod
-            # fping 一次性 ping 多个 IP
+    async def _load_recent_valid(
-            # -c 1: 每个 IP 发 1 个包
+        db: AsyncSession,
-            # -t: 超时毫秒
+        device_id: int,
-            timeout_ms = int(settings.PING_TIMEOUT_SECONDS * 1000)
+        policy: DeviceMonitoringPolicy,
-            cmd = [
+    ) -> list[PingRecord]:
-                settings.FPING_PATH,
+        """Load enough valid summaries for every configured transition rule."""
-                "-c", "1",
+        history_size = max(
-                "-t", str(timeout_ms),
+            policy.offline_consecutive_rounds,
-                "-e",  # 显示响应时间
+            policy.degraded_window_rounds,
-            ] + ip_list
+            policy.recovery_consecutive_clean_rounds,
+        )
+        query_result = await db.execute(
+            select(PingRecord)
+            .where(PingRecord.device_id == device_id)
+            .where(PingRecord.is_valid.is_(True))
+            .order_by(PingRecord.created_at.desc(), PingRecord.id.desc())
+            .limit(history_size)
+        )
+        newest_first = list(query_result.scalars().all())
+        return list(reversed(newest_first))
 
-            proc = await asyncio.create_subprocess_exec(
+    @staticmethod
-                *cmd,
+    def _update_observation_fields(
-                stdout=subprocess.PIPE,
+        device: Device,
-                stderr=subprocess.PIPE,
+        recent: list[PingRecord],
-            )
+        observed_at: datetime,
-            stdout, stderr = await proc.communicate()
+    ) -> None:
+        """Maintain compatibility timestamps and the full-loss streak counter."""
+        current = recent[-1]
+        if current.received_count > 0:
+            device.last_online_time = observed_at
 
-            result_map: dict[str, tuple[bool, Optional[float]]] = {}
+        full_loss_count = 0
+        for record in reversed(recent):
+            if record.sent_count and record.received_count == 0:
+                full_loss_count += 1
+            else:
+                break
+        device.consecutive_failures = full_loss_count
 
-            # fping 标准输出逐行: "IP : xmt/rcv/%loss = 1/1/0%, rtt min/avg/max = 0.12/0.12/0.12"
+    @staticmethod
-            # 或 "IP : xmt/rcv/%loss = 1/0/100%"
+    def _apply_decision(
-            for line in stdout.decode("utf-8", errors="replace").splitlines():
+        device: Device,
-                line = line.strip()
+        decision: StateDecision,
-                if ":" not in line:
+        observed_at: datetime,
-                    continue
+    ) -> DeviceStateChange | None:
-                ip = line.split(":")[0].strip()
+        """Apply one pure decision and materialize its transition value."""
-                # 解析响应时间
+        old_status = device.current_status or "unknown"
-                if "rtt" in line:
+        if decision.next_status == old_status:
-                    try:
+            return None
-                        # 提取 avg rtt
-                        rtt_part = line.split("rtt")[1]
-                        # 格式: min/avg/max = 0.12/0.12/0.12
-                        if "=" in rtt_part:
-                            avg_rtt_str = rtt_part.split("=")[1].strip().split("/")[1]
-                            rtt_ms = float(avg_rtt_str)
-                        else:
-                            rtt_ms = None
-                    except (IndexError, ValueError):
-                        rtt_ms = None
-                    result_map[ip] = (True, rtt_ms)
-                else:
-                    result_map[ip] = (False, None)
 
-            return result_map
+        device.current_status = decision.next_status
+        if decision.next_status == "offline":
+            device.last_offline_time = observed_at
+        if decision.next_status == "online":
+            device.last_online_time = observed_at
 
-        except FileNotFoundError:
+        return DeviceStateChange(
-            logger.warning("fping 未找到，回退到系统 ping (串行)")
+            device=device,
-            return await self._fallback_ping(ip_list)
+            old_status=old_status,
-        except Exception as e:
+            new_status=decision.next_status,
-            logger.error(f"fping 异常: {e}")
+            consecutive_failures=device.consecutive_failures or 0,
-            return await self._fallback_ping(ip_list)
+            event_type=decision.event_type,
-
+            reason=decision.reason,
-    async def _fallback_ping(self, ip_list: list[str]) -> dict[str, tuple[bool, Optional[float]]]:
+        )
-        """回退方案：使用系统 ping，并发执行"""
-        async def ping_one(ip: str) -> tuple[str, bool, Optional[float]]:
-            try:
-                timeout = settings.PING_TIMEOUT_SECONDS
-                cmd = ["ping", "-c", "1", "-W", str(int(timeout)), ip]
-                start = time.time()
-                proc = await asyncio.create_subprocess_exec(
-                    *cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-                )
-                await proc.wait()
-                elapsed = (time.time() - start) * 1000
-                return ip, proc.returncode == 0, round(elapsed, 2)
-            except Exception:
-                return ip, False, None
-
-        tasks = [ping_one(ip) for ip in ip_list]
-        sem = asyncio.Semaphore(settings.PING_CONCURRENCY)
-
-        async def bounded_ping(ip: str):
-            async with sem:
-                return await ping_one(ip)
-
-        results = await asyncio.gather(*[bounded_ping(ip) for ip in ip_list])
-        return {ip: (alive, rtt) for ip, alive, rtt in results}

backend/app/services/scheduler.py

@@ -1,94 +1,142 @@
-"""
+"""Single-task scheduler for probe, event recording, and outbox delivery."""
-定时任务调度器
-
-使用 asyncio 循环驱动 Ping 引擎，协调 Pinger 和 Alerter。
-"""
 
 import asyncio
 import logging
+import time
 from datetime import datetime
+from typing import Callable
 
-from sqlalchemy.ext.asyncio import AsyncSession
+from app.config import Settings, settings
-
-from app.config import settings
-from app.services.pinger import Pinger
-from app.services.alerter import Alerter
 from app.core.deps import async_session
+from app.services.alerter import Alerter
+from app.services.notification_dispatcher import (
+    NotificationDispatcher,
+    WeComClient,
+)
+from app.services.pinger import Pinger
+
 
 logger = logging.getLogger("pingwatch.scheduler")
+_DEFAULT_DISPATCH_INTERVAL_SECONDS = 5.0
 
 
 class PingScheduler:
-    """
+    """Serialize monitoring cycles and retain one cancellable background task."""
-    调度器职责：
-    1. 按间隔驱动 Ping 引擎
-    2. 每轮结束后触发 Alerter 处理待发送告警
-    3. 控制并发和清理
-    """
 
-    def __init__(self):
+    def __init__(
-        self._pinger = Pinger()
+        self,
-        self._alerter = Alerter()
+        *,
+        pinger=None,
+        alerter=None,
+        dispatcher=None,
+        session_factory: Callable = async_session,
+        runtime_settings: Settings | None = None,
+        interval_seconds: float | None = None,
+        dispatch_interval_seconds: float = _DEFAULT_DISPATCH_INTERVAL_SECONDS,
+    ):
+        self._settings = runtime_settings or settings
+        self._pinger = pinger or Pinger(self._settings)
+        self._alerter = alerter or Alerter(self._settings)
+        self._notification_client = None
+        self._dispatch_enabled = (
+            dispatcher is not None or self._settings.wecom_notification_enabled
+        )
+        if dispatcher is None:
+            self._notification_client = WeComClient(self._settings)
+            dispatcher = NotificationDispatcher(
+                self._notification_client,
+                self._settings,
+            )
+        self._dispatcher = dispatcher
+        self._session_factory = session_factory
+        self._probe_interval_seconds = max(
+            1.0,
+            float(
+                interval_seconds
+                if interval_seconds is not None
+                else self._settings.PING_INTERVAL_SECONDS
+            ),
+        )
+        self._dispatch_interval_seconds = min(
+            max(1.0, float(dispatch_interval_seconds)),
+            30.0,
+        )
         self._running = False
         self._task: asyncio.Task | None = None
+        self._cycle_lock = asyncio.Lock()
 
-        # 注册状态变化回调
+    async def _run_cycle(self, *, run_probe: bool = True) -> None:
-        self._pinger.on_state_change(self._on_state_change)
+        """Run one serialized transaction cycle and close its session on cancel."""
+        async with self._cycle_lock:
+            async with self._session_factory() as db:
+                try:
+                    if run_probe:
+                        changes = await self._pinger.run_one_round(db)
+                        for change in changes:
+                            await self._alerter.record_transition(change, db)
+                        await db.commit()
 
-    async def _on_state_change(self, change):
+                    if self._dispatch_enabled:
-        """收到设备状态变化，转给 alerter"""
+                        await self._dispatcher.dispatch_due(db, datetime.now())
-        async with async_session() as db:
+                        await db.commit()
-            try:
+                except BaseException:
-                await self._alerter.on_state_change(change, db)
+                    await db.rollback()
-            except Exception as e:
+                    raise
-                logger.error(f"告警处理异常: {e}", exc_info=True)
-
-    async def _run_loop(self):
-        """主循环"""
-        logger.info("Ping 调度器已启动")
-        self._running = True
 
+    async def _run_loop(self) -> None:
+        """Probe at its configured cadence and service retries every few seconds."""
+        logger.info("Ping scheduler started")
+        next_probe_at = 0.0
         while self._running:
+            now = time.monotonic()
+            run_probe = now >= next_probe_at
             try:
-                async with async_session() as db:
+                await self._run_cycle(run_probe=run_probe)
-                    # 执行一轮 ping
+                if run_probe:
-                    results = await self._pinger.run_one_round(db)
+                    next_probe_at = time.monotonic() + self._probe_interval_seconds
-
-                    if results:
-                        # 直接用启用的设备数
-                        total = len(results)
-
-                        # 处理待发送告警
-                        await self._alerter.flush_pending(db, total)
-
             except asyncio.CancelledError:
                 break
-            except Exception as e:
+            except Exception as exc:
-                logger.error(f"调度器异常: {e}", exc_info=True)
+                logger.error(
+                    "Scheduler cycle failed type=%s",
+                    type(exc).__name__,
+                    exc_info=True,
+                )
+                if run_probe:
+                    next_probe_at = time.monotonic() + self._probe_interval_seconds
 
-            # 等待下一轮
+            seconds_until_probe = max(0.0, next_probe_at - time.monotonic())
-            await asyncio.sleep(settings.PING_INTERVAL_SECONDS)
+            sleep_seconds = min(
+                self._dispatch_interval_seconds,
+                seconds_until_probe or self._dispatch_interval_seconds,
+            )
+            try:
+                await asyncio.sleep(sleep_seconds)
+            except asyncio.CancelledError:
+                break
+        logger.info("Ping scheduler stopped")
 
-        logger.info("Ping 调度器已停止")
+    def start(self) -> None:
-
+        """Start exactly one background scheduler task."""
-    def start(self):
+        if self._task is not None and not self._task.done():
-        """启动调度器（后台任务）"""
+            logger.warning("Ping scheduler is already running")
-        if self._running:
-            logger.warning("调度器已在运行")
             return
+        self._running = True
         self._task = asyncio.create_task(self._run_loop())
 
-    async def stop(self):
+    async def stop(self) -> None:
-        """停止调度器"""
+        """Cancel and await the active cycle before releasing its HTTP client."""
         self._running = False
-        if self._task:
+        if self._task is not None:
             self._task.cancel()
             try:
                 await self._task
             except asyncio.CancelledError:
                 pass
             self._task = None
+        if self._notification_client is not None:
+            await self._notification_client.aclose()
+            self._notification_client = None
 
 
-# 全局调度器实例
 scheduler = PingScheduler()

backend/app/services/settings_validation.py

@@ -0,0 +1,21 @@
+"""Runtime validation for security-sensitive application settings."""
+
+from app.config import LEGACY_DEFAULT_SECRET, Settings
+
+
+def validate_runtime_settings(settings: Settings) -> None:
+    """Reject unsafe settings before scheduling monitoring work."""
+    if settings.environment == "production":
+        if not settings.secret_key.strip() or settings.secret_key == LEGACY_DEFAULT_SECRET:
+            raise ValueError("SECRET_KEY must be provided by the runtime environment")
+        if not settings.casdoor_endpoint.startswith("https://"):
+            raise ValueError("CASDOOR endpoint must use TLS in production")
+
+    if settings.wecom_notification_enabled:
+        credentials = (
+            settings.WECOM_CORP_ID,
+            settings.WECOM_AGENT_ID,
+            settings.WECOM_APP_SECRET,
+        )
+        if not all(credentials):
+            raise ValueError("WECOM delivery requires corp ID, agent ID, and app secret")

backend/app/services/state_machine.py

@@ -0,0 +1,126 @@
+"""Pure state transitions for device connectivity health."""
+
+from dataclasses import dataclass
+from typing import Protocol, Sequence
+
+from app.models.device import DeviceMonitoringPolicy
+
+
+class ProbeSample(Protocol):
+    """Packet summary attributes required for health evaluation."""
+
+    sent_count: int
+    received_count: int
+    is_valid: bool
+
+
+@dataclass(frozen=True)
+class StateDecision:
+    """Result of evaluating recent probes against a monitoring policy."""
+
+    next_status: str
+    event_type: str | None
+    should_notify: bool
+    reason: str
+
+
+def _retain(previous_status: str, reason: str) -> StateDecision:
+    """Return a non-notifying decision that preserves current health."""
+    return StateDecision(
+        next_status=previous_status,
+        event_type=None,
+        should_notify=False,
+        reason=reason,
+    )
+
+
+def _transition(
+    previous_status: str,
+    next_status: str,
+    event_type: str,
+    reason: str,
+) -> StateDecision:
+    """Return a transition event only when the persisted status changes."""
+    if previous_status == next_status:
+        return _retain(previous_status, reason)
+    return StateDecision(
+        next_status=next_status,
+        event_type=event_type,
+        should_notify=True,
+        reason=reason,
+    )
+
+
+def _is_valid(sample: ProbeSample) -> bool:
+    """Reject impossible summaries before they influence device health."""
+    return (
+        sample.is_valid
+        and sample.sent_count > 0
+        and 0 <= sample.received_count <= sample.sent_count
+    )
+
+
+def evaluate_health(
+    previous_status: str,
+    recent: Sequence[ProbeSample],
+    policy: DeviceMonitoringPolicy,
+) -> StateDecision:
+    """Evaluate ordered oldest-to-newest probe summaries.
+
+    Invalid input retains state. Offline has priority over aggregate degraded
+    loss, and an alerted state recovers only after the configured clean streak.
+    """
+    if not recent:
+        return _retain(previous_status, "no probe history")
+    if any(not _is_valid(sample) for sample in recent):
+        return _retain(previous_status, "invalid probe history")
+
+    offline_rounds = policy.offline_consecutive_rounds
+    if len(recent) >= offline_rounds and all(
+        sample.received_count == 0
+        for sample in recent[-offline_rounds:]
+    ):
+        return _transition(
+            previous_status,
+            "offline",
+            "offline",
+            f"{offline_rounds} consecutive full-loss rounds",
+        )
+
+    recovery_rounds = policy.recovery_consecutive_clean_rounds
+    if (
+        previous_status in {"offline", "degraded"}
+        and len(recent) >= recovery_rounds
+        and all(
+            sample.received_count == sample.sent_count
+            for sample in recent[-recovery_rounds:]
+        )
+    ):
+        return _transition(
+            previous_status,
+            "online",
+            "recovered",
+            f"{recovery_rounds} consecutive clean rounds",
+        )
+
+    degraded_rounds = policy.degraded_window_rounds
+    if len(recent) >= degraded_rounds:
+        window = recent[-degraded_rounds:]
+        sent_count = sum(sample.sent_count for sample in window)
+        received_count = sum(sample.received_count for sample in window)
+        loss_percent = (sent_count - received_count) / sent_count * 100.0
+        if (
+            previous_status != "offline"
+            and loss_percent >= policy.degraded_loss_percent
+        ):
+            return _transition(
+                previous_status,
+                "degraded",
+                "degraded",
+                (
+                    f"window packet loss {loss_percent:.2f}% is at or above "
+                    f"{policy.degraded_loss_percent:.2f}%"
+                ),
+            )
+
+    return _retain(previous_status, "no transition threshold met")

backend/migrations/versions/20260803_reliability_monitoring.py

@@ -0,0 +1,152 @@
+"""Add durable reliability-monitoring persistence without destructive DDL."""
+
+from datetime import datetime
+
+from sqlalchemy import Column, DateTime, MetaData, String, Table, inspect, select, text
+from sqlalchemy.engine import Connection
+
+from app.models import AlertEvent, Device, NotificationOutbox, PingRecord
+
+
+REVISION = "20260803_reliability_monitoring"
+
+_migration_metadata = MetaData()
+_schema_migrations = Table(
+    "schema_migrations",
+    _migration_metadata,
+    Column("revision", String(64), primary_key=True),
+    Column("applied_at", DateTime, nullable=False, default=datetime.now),
+)
+
+_DEFAULTS_FOR_EXISTING_ROWS = {
+    "is_valid": "TRUE",
+    "notification_attempts": "0",
+}
+
+_REVISION_COLUMNS = {
+    "devices": (
+        Device.__table__,
+        {
+            "probe_packets_per_round",
+            "offline_consecutive_rounds",
+            "degraded_window_rounds",
+            "degraded_loss_percent",
+            "recovery_consecutive_clean_rounds",
+        },
+    ),
+    "ping_records": (
+        PingRecord.__table__,
+        {
+            "sent_count",
+            "received_count",
+            "packet_loss_percent",
+            "average_rtt_ms",
+            "is_valid",
+            "failure_reason",
+        },
+    ),
+    "alert_events": (
+        AlertEvent.__table__,
+        {
+            "notification_attempts",
+            "last_notification_error",
+            "related_event_id",
+            "previous_status",
+            "current_status",
+        },
+    ),
+}
+
+
+def _quote(connection: Connection, identifier: str) -> str:
+    """Quote one database identifier through the active dialect."""
+    return connection.dialect.identifier_preparer.quote(identifier)
+
+
+def _add_missing_columns(connection: Connection) -> None:
+    """Add only newly required nullable/defaulted columns to existing tables."""
+    inspector = inspect(connection)
+    existing_tables = set(inspector.get_table_names())
+    for table_name, (table, column_names) in _REVISION_COLUMNS.items():
+        if table_name not in existing_tables:
+            continue
+
+        existing_columns = {
+            column["name"] for column in inspector.get_columns(table_name)
+        }
+        for column in table.columns:
+            if column.name not in column_names or column.name in existing_columns:
+                continue
+
+            column_type = connection.dialect.type_compiler.process(column.type)
+            definition = f"{_quote(connection, column.name)} {column_type}"
+            default = _DEFAULTS_FOR_EXISTING_ROWS.get(column.name)
+            if default is not None:
+                if connection.dialect.name == "sqlite" and default == "TRUE":
+                    default = "1"
+                if column.name == "is_valid":
+                    definition = f"{definition} NOT NULL DEFAULT {default}"
+                else:
+                    definition = f"{definition} DEFAULT {default}"
+            connection.execute(
+                text(
+                    f"ALTER TABLE {_quote(connection, table_name)} "
+                    f"ADD COLUMN {definition}"
+                )
+            )
+
+
+def _create_missing_indexes(connection: Connection) -> None:
+    """Create new performance indexes for tables that predate this revision."""
+    inspector = inspect(connection)
+    table = PingRecord.__table__
+    if table.name not in inspector.get_table_names():
+        return
+    existing_indexes = {
+        index["name"] for index in inspector.get_indexes(table.name)
+    }
+    index = next(
+        item
+        for item in table.indexes
+        if item.name == "idx_ping_records_device_created_at"
+    )
+    if index.name not in existing_indexes:
+        index.create(bind=connection)
+
+
+def _upgrade_postgresql_alert_type(connection: Connection) -> None:
+    """Permit the degraded alert value for databases using PostgreSQL enums."""
+    if connection.dialect.name == "postgresql":
+        enum_name = AlertEvent.__table__.c.alert_type.type.name
+        connection.execute(
+            text(
+                f"ALTER TYPE {_quote(connection, enum_name)} "
+                "ADD VALUE IF NOT EXISTS 'degraded'"
+            )
+        )
+
+
+def run_reliability_migration(connection: Connection) -> None:
+    """Apply this revision once while retaining all existing monitoring data.
+
+    This runner is deliberately idempotent: it creates only this revision's
+    outbox table, missing revision columns, and indexes, then writes one
+    immutable revision marker. It never issues DROP, DELETE, UPDATE, or
+    data-copy statements. The application owns global metadata creation after
+    this function returns.
+    """
+    _migration_metadata.create_all(bind=connection)
+    NotificationOutbox.__table__.create(bind=connection, checkfirst=True)
+
+    already_applied = connection.execute(
+        select(_schema_migrations.c.revision).where(
+            _schema_migrations.c.revision == REVISION
+        )
+    ).scalar_one_or_none()
+    if already_applied is not None:
+        return
+
+    _upgrade_postgresql_alert_type(connection)
+    _add_missing_columns(connection)
+    _create_missing_indexes(connection)
+    connection.execute(_schema_migrations.insert().values(revision=REVISION))

backend/pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+asyncio_mode = auto
+asyncio_default_fixture_loop_scope = function
+testpaths = tests

backend/requirements.txt

@@ -1,7 +1,7 @@
 fastapi>=0.110.0
 uvicorn[standard]>=0.29.0
 sqlalchemy[asyncio]>=2.0.30
-aiosqlite>=0.20.0
+aiosqlite==0.20.0
 asyncpg>=0.29.0
 pydantic>=2.7.0
 pydantic-settings>=2.2.0
@@ -9,3 +9,5 @@ python-jose[cryptography]>=3.3.0
 httpx>=0.27.0
 python-multipart>=0.0.9
 websockets>=12.0
+pytest==8.3.5
+pytest-asyncio==0.25.3

backend/tests/conftest.py

@@ -0,0 +1 @@
+"""Shared test configuration for PingWatch backend tests."""

backend/tests/test_alert_workflow.py

@@ -0,0 +1,427 @@
+"""Transactional alert-event and notification-outbox workflow tests."""
+
+import asyncio
+from datetime import datetime, timedelta
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import (
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+
+from app.config import Settings
+from app.models import (
+    AlertEvent,
+    AlertTypeEnum,
+    Base,
+    Device,
+    DeviceTypeEnum,
+    NotificationOutbox,
+    NotificationStatus,
+    PingRecord,
+)
+from app.services.alerter import Alerter
+from app.services.fping_runner import ProbeResult
+from app.services.pinger import DeviceStateChange, Pinger
+from app.services.scheduler import PingScheduler
+
+
+@pytest.fixture
+async def db_session() -> AsyncSession:
+    """Provide a complete isolated persistence boundary for alert tests."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            yield session
+    finally:
+        await engine.dispose()
+
+
+@pytest.fixture
+async def offline_change(db_session: AsyncSession) -> DeviceStateChange:
+    """Persist a device and expose a real offline state-change value."""
+    observed_at = datetime(2026, 8, 4, 9, 30)
+    device = Device(
+        name="核心交换机",
+        ip="10.0.0.8",
+        device_type=DeviceTypeEnum.switch,
+        location="一楼机房",
+        project_name="园区网",
+        current_status="offline",
+        last_ping_time=observed_at,
+    )
+    db_session.add(device)
+    await db_session.commit()
+    return DeviceStateChange(
+        device=device,
+        old_status="online",
+        new_status="offline",
+        consecutive_failures=2,
+        event_type="offline",
+        reason="2 consecutive full-loss rounds",
+    )
+
+
+async def test_transition_creates_event_and_pending_outbox_in_one_transaction(
+    db_session: AsyncSession,
+    offline_change: DeviceStateChange,
+):
+    """Removing either insert breaks the durable event-delivery contract."""
+    event = await Alerter().record_transition(offline_change, db_session)
+
+    outbox = await db_session.scalar(
+        select(NotificationOutbox).where(
+            NotificationOutbox.alert_event_id == event.id
+        )
+    )
+
+    assert outbox is not None
+    assert outbox.status == NotificationStatus.pending
+    assert "核心交换机" in outbox.message_content
+    assert "10.0.0.8" in outbox.message_content
+    assert "一楼机房" in outbox.message_content
+    assert "园区网" in outbox.message_content
+    assert "2 consecutive full-loss rounds" in outbox.message_content
+    assert "2026-08-04 09:30:00" in outbox.message_content
+
+
+async def test_event_and_outbox_rollback_together(
+    db_session: AsyncSession,
+    offline_change: DeviceStateChange,
+):
+    """A caller rollback cannot retain an event without its notification."""
+    await Alerter().record_transition(offline_change, db_session)
+
+    await db_session.rollback()
+
+    assert await db_session.scalar(select(AlertEvent)) is None
+    assert await db_session.scalar(select(NotificationOutbox)) is None
+
+
+async def test_offline_escalation_closes_and_links_open_degraded_event(
+    db_session: AsyncSession,
+    offline_change: DeviceStateChange,
+):
+    """Escalating degraded to offline must not leave two open incidents."""
+    device = offline_change.device
+    device.last_ping_time = datetime(2026, 8, 4, 9, 20)
+    degraded = await Alerter().record_transition(
+        DeviceStateChange(
+            device=device,
+            old_status="online",
+            new_status="degraded",
+            consecutive_failures=0,
+            event_type="degraded",
+            reason="window packet loss 20.00%",
+        ),
+        db_session,
+    )
+    await db_session.commit()
+    device.last_ping_time = datetime(2026, 8, 4, 9, 30)
+
+    offline = await Alerter().record_transition(
+        DeviceStateChange(
+            device=device,
+            old_status="degraded",
+            new_status="offline",
+            consecutive_failures=2,
+            event_type="offline",
+            reason="2 consecutive full-loss rounds",
+        ),
+        db_session,
+    )
+
+    assert degraded.is_resolved is True
+    assert degraded.end_at == datetime(2026, 8, 4, 9, 30)
+    assert degraded.duration_minutes == 10
+    assert degraded.related_event_id == offline.id
+    assert offline.related_event_id == degraded.id
+
+
+async def test_recovery_closes_open_fault_and_notifies_with_duration(
+    db_session: AsyncSession,
+    offline_change: DeviceStateChange,
+):
+    """Recovery closes one active incident and carries its duration to operators."""
+    opened = await Alerter().record_transition(offline_change, db_session)
+    await db_session.commit()
+    device = offline_change.device
+    device.last_ping_time = datetime(2026, 8, 4, 10, 1)
+
+    recovered = await Alerter().record_transition(
+        DeviceStateChange(
+            device=device,
+            old_status="offline",
+            new_status="online",
+            consecutive_failures=0,
+            event_type="recovered",
+            reason="3 consecutive clean rounds",
+        ),
+        db_session,
+    )
+    outbox = await db_session.scalar(
+        select(NotificationOutbox).where(
+            NotificationOutbox.alert_event_id == recovered.id
+        )
+    )
+
+    assert opened.is_resolved is True
+    assert opened.end_at == datetime(2026, 8, 4, 10, 1)
+    assert opened.duration_minutes == 31
+    assert recovered.alert_type == AlertTypeEnum.recovered
+    assert recovered.is_resolved is True
+    assert recovered.duration_minutes == 31
+    assert recovered.related_event_id == opened.id
+    assert "持续时间：31 分钟" in outbox.message_content
+
+
+async def test_probe_state_event_and_outbox_share_the_caller_transaction(
+    monkeypatch,
+):
+    """A failed caller commit cannot persist a state transition without its event."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    observed_at = datetime.now() - timedelta(seconds=30)
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        async with AsyncSession(engine, expire_on_commit=False) as setup:
+            device = Device(
+                name="edge-atomic",
+                ip="10.0.0.18",
+                device_type=DeviceTypeEnum.switch,
+                current_status="online",
+                is_enabled=True,
+                offline_consecutive_rounds=2,
+            )
+            setup.add(device)
+            await setup.flush()
+            device_id = device.id
+            setup.add(
+                PingRecord(
+                    device_id=device_id,
+                    is_alive=False,
+                    response_time_ms=None,
+                    round_num=1,
+                    sent_count=3,
+                    received_count=0,
+                    packet_loss_percent=100.0,
+                    average_rtt_ms=None,
+                    is_valid=True,
+                    created_at=observed_at,
+                )
+            )
+            await setup.commit()
+
+        async def fake_run_fping_count(*args, **kwargs):
+            return {
+                "10.0.0.18": ProbeResult(
+                    "10.0.0.18",
+                    3,
+                    0,
+                    None,
+                    True,
+                )
+            }
+
+        monkeypatch.setattr(
+            "app.services.pinger.run_fping_count",
+            fake_run_fping_count,
+        )
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            changes = await Pinger().run_one_round(session)
+            await Alerter().record_transition(changes[0], session)
+            await session.rollback()
+
+        async with AsyncSession(engine, expire_on_commit=False) as verification:
+            persisted_device = await verification.get(Device, device_id)
+            records = list(
+                (
+                    await verification.execute(
+                        select(PingRecord).where(PingRecord.device_id == device_id)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert persisted_device.current_status == "online"
+            assert len(records) == 1
+            assert await verification.scalar(select(AlertEvent)) is None
+            assert await verification.scalar(select(NotificationOutbox)) is None
+    finally:
+        await engine.dispose()
+
+
+class RecordingSession:
+    """Small session boundary exposing transaction and close ordering."""
+
+    def __init__(self, events: list[str]):
+        self.events = events
+        self.closed = asyncio.Event()
+
+    async def __aenter__(self):
+        self.events.append("session-enter")
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        self.events.append("session-exit")
+        self.closed.set()
+
+    async def commit(self):
+        self.events.append("commit")
+
+    async def rollback(self):
+        self.events.append("rollback")
+
+
+class RecordingPinger:
+    def __init__(self, events: list[str], changes=None):
+        self.events = events
+        self.changes = list(changes or ["transition"])
+
+    async def run_one_round(self, db):
+        self.events.append("probe")
+        return self.changes
+
+
+class RecordingAlerter:
+    def __init__(self, events: list[str]):
+        self.events = events
+
+    async def record_transition(self, change, db):
+        self.events.append(f"alert:{change}")
+
+
+class RecordingDispatcher:
+    def __init__(self, events: list[str]):
+        self.events = events
+
+    async def dispatch_due(self, db, now):
+        assert isinstance(now, datetime)
+        self.events.append("dispatch")
+
+
+async def test_scheduler_persists_transitions_before_dispatching_due_messages():
+    """Changing scheduler order cannot expose uncommitted outbox rows to dispatch."""
+    events: list[str] = []
+    session = RecordingSession(events)
+    scheduler = PingScheduler(
+        pinger=RecordingPinger(events),
+        alerter=RecordingAlerter(events),
+        dispatcher=RecordingDispatcher(events),
+        session_factory=lambda: session,
+    )
+
+    await scheduler._run_cycle()
+
+    assert events == [
+        "session-enter",
+        "probe",
+        "alert:transition",
+        "commit",
+        "dispatch",
+        "commit",
+        "session-exit",
+    ]
+
+
+async def test_scheduler_lock_prevents_overlapping_cycles():
+    """Even concurrent triggers cannot overlap probe or notification sessions."""
+    events: list[str] = []
+
+    class ConcurrencyPinger:
+        active = 0
+        maximum = 0
+
+        async def run_one_round(self, db):
+            self.active += 1
+            self.maximum = max(self.maximum, self.active)
+            await asyncio.sleep(0.01)
+            self.active -= 1
+            return []
+
+    pinger = ConcurrencyPinger()
+    scheduler = PingScheduler(
+        pinger=pinger,
+        alerter=RecordingAlerter(events),
+        dispatcher=RecordingDispatcher(events),
+        session_factory=lambda: RecordingSession(events),
+    )
+
+    await asyncio.gather(scheduler._run_cycle(), scheduler._run_cycle())
+
+    assert pinger.maximum == 1
+
+
+async def test_scheduler_stop_waits_for_active_session_to_close():
+    """Cancellation cannot return while a probe session remains open."""
+    events: list[str] = []
+    session = RecordingSession(events)
+    probe_started = asyncio.Event()
+    never_complete = asyncio.Event()
+
+    class BlockingPinger:
+        async def run_one_round(self, db):
+            probe_started.set()
+            await never_complete.wait()
+
+    scheduler = PingScheduler(
+        pinger=BlockingPinger(),
+        alerter=RecordingAlerter(events),
+        dispatcher=RecordingDispatcher(events),
+        session_factory=lambda: session,
+        interval_seconds=1,
+    )
+    scheduler.start()
+    first_task = scheduler._task
+    scheduler.start()
+
+    await probe_started.wait()
+    await scheduler.stop()
+
+    assert first_task is not None
+    assert session.closed.is_set()
+    assert scheduler._task is None
+    assert events[-2:] == ["rollback", "session-exit"]
+
+
+async def test_scheduler_leaves_pending_outbox_untouched_when_delivery_disabled():
+    """Disabling WeCom preserves queued notifications for a later enablement."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    session_factory = async_sessionmaker(engine, expire_on_commit=False)
+
+    class EmptyPinger:
+        async def run_one_round(self, db):
+            return []
+
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        async with session_factory() as setup:
+            setup.add(
+                NotificationOutbox(
+                    alert_event_id=42,
+                    message_content="保留待发送事件",
+                )
+            )
+            await setup.commit()
+
+        runtime_settings = Settings(wecom_notification_enabled=False)
+        scheduler = PingScheduler(
+            pinger=EmptyPinger(),
+            alerter=Alerter(runtime_settings),
+            session_factory=session_factory,
+            runtime_settings=runtime_settings,
+        )
+        await scheduler._run_cycle()
+
+        async with session_factory() as verification:
+            outbox = await verification.scalar(select(NotificationOutbox))
+            assert outbox.status == NotificationStatus.pending
+            assert outbox.attempt_count == 0
+        await scheduler.stop()
+    finally:
+        await engine.dispose()

backend/tests/test_deployment_security.py

@@ -0,0 +1,18 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_compose_does_not_embed_password_or_net_admin_capability():
+    compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
+
+    assert "POSTGRES_PASSWORD=pingwatch123" not in compose
+    assert "NET_ADMIN" not in compose
+    assert "127.0.0.1:8001:8000" in compose
+
+
+def test_oauth_client_does_not_disable_tls_verification():
+    auth_source = (ROOT / "backend" / "app" / "core" / "auth.py").read_text(encoding="utf-8")
+
+    assert "verify=False" not in auth_source

backend/tests/test_deps_import.py

@@ -0,0 +1,48 @@
+"""Regression coverage for application dependency module imports."""
+
+import importlib
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def test_dependency_module_compiles():
+    """Database initialization dependencies remain valid Python syntax."""
+    module_path = Path(__file__).parents[1] / "app" / "core" / "deps.py"
+
+    compile(module_path.read_text(encoding="utf-8"), str(module_path), "exec")
+
+
+@pytest.mark.parametrize(
+    ("database_url", "expected_url"),
+    [
+        ("sqlite:///./pingwatch.db", "sqlite+aiosqlite:///./pingwatch.db"),
+        (
+            "sqlite+aiosqlite:///./pingwatch.db",
+            "sqlite+aiosqlite:///./pingwatch.db",
+        ),
+    ],
+)
+def test_dependency_module_normalizes_sqlite_url_once(
+    monkeypatch, database_url, expected_url
+):
+    """Plain SQLite URLs gain the async driver without duplicating an existing one."""
+    monkeypatch.setenv("DATABASE_URL", database_url)
+    sys.modules.pop("app.core.deps", None)
+    sys.modules.pop("app.config", None)
+
+    deps = importlib.import_module("app.core.deps")
+
+    assert str(deps.engine.url) == expected_url
+
+
+def test_dependency_module_initializes_with_default_sqlite_url(monkeypatch):
+    """The default database setting can initialize the asynchronous dependency."""
+    monkeypatch.delenv("DATABASE_URL", raising=False)
+    sys.modules.pop("app.core.deps", None)
+    sys.modules.pop("app.config", None)
+
+    deps = importlib.import_module("app.core.deps")
+
+    assert str(deps.engine.url) == "sqlite+aiosqlite:///./pingwatch.db"

backend/tests/test_fping_runner.py

@@ -0,0 +1,322 @@
+"""Behavior tests for multi-packet fping execution and persistence."""
+
+import asyncio
+from datetime import datetime, timedelta
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+from app.config import Settings
+from app.models.device import Base, Device, DeviceTypeEnum
+from app.models.ping_record import PingRecord
+from app.services.fping_runner import (
+    ProbeResult,
+    parse_fping_count_output,
+    run_fping_count,
+)
+from app.services.pinger import Pinger
+
+
+@pytest.fixture
+async def db_session() -> AsyncSession:
+    """Provide an isolated database for pinger transaction tests."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            yield session
+    finally:
+        await engine.dispose()
+
+
+def test_parses_loss_and_average_rtt():
+    """Numeric replies count as received packets and dashes count as loss."""
+    output = "10.0.0.8 : 12.4 - 12.6\n10.0.0.9 : - - -"
+
+    result = parse_fping_count_output(
+        output,
+        {"10.0.0.8", "10.0.0.9"},
+        3,
+    )
+
+    assert result["10.0.0.8"].received_count == 2
+    assert result["10.0.0.8"].average_rtt_ms == 12.5
+    assert result["10.0.0.8"].packet_loss_percent == pytest.approx(
+        33.3333333333
+    )
+    assert result["10.0.0.9"].packet_loss_percent == 100.0
+
+
+def test_marks_missing_and_malformed_expected_output_invalid():
+    """Incomplete fping output cannot be interpreted as device packet loss."""
+    output = "10.0.0.8 : 10.2 bad-token -"
+
+    result = parse_fping_count_output(
+        output,
+        {"10.0.0.8", "10.0.0.9"},
+        3,
+    )
+
+    assert result["10.0.0.8"].is_valid is False
+    assert result["10.0.0.8"].failure_reason == "malformed fping replies"
+    assert result["10.0.0.9"].is_valid is False
+    assert result["10.0.0.9"].failure_reason == "missing fping output"
+
+
+def test_parses_ipv6_without_confusing_address_colons_for_delimiter():
+    """An IPv6 address remains the result key when parsing the output delimiter."""
+    result = parse_fping_count_output(
+        "2001:db8::8 : 1.25 - 1.75",
+        {"2001:db8::8"},
+        3,
+    )
+
+    assert result["2001:db8::8"].is_valid is True
+    assert result["2001:db8::8"].received_count == 2
+    assert result["2001:db8::8"].average_rtt_ms == 1.5
+
+
+async def test_runner_uses_argument_vector_and_parses_quiet_stderr(monkeypatch):
+    """The runner passes IPs as argv entries and reads fping -C quiet output."""
+    observed = {}
+
+    class CompletedProcess:
+        returncode = 1
+
+        async def communicate(self):
+            return b"", b"10.0.0.8 : 4.2 - 4.4\n"
+
+    async def fake_create_subprocess_exec(*args, **kwargs):
+        observed["args"] = args
+        observed["kwargs"] = kwargs
+        return CompletedProcess()
+
+    monkeypatch.setattr(
+        asyncio,
+        "create_subprocess_exec",
+        fake_create_subprocess_exec,
+    )
+
+    result = await run_fping_count(
+        ["10.0.0.8"],
+        packets_per_round=3,
+        timeout_ms=800,
+        executable="/test/fping",
+    )
+
+    assert observed["args"] == (
+        "/test/fping",
+        "-C",
+        "3",
+        "-q",
+        "-t",
+        "800",
+        "10.0.0.8",
+    )
+    assert observed["kwargs"] == {
+        "stdout": asyncio.subprocess.PIPE,
+        "stderr": asyncio.subprocess.PIPE,
+    }
+    assert result["10.0.0.8"].received_count == 2
+
+
+async def test_runner_failure_returns_invalid_result_instead_of_packet_loss(
+    monkeypatch,
+):
+    """A process communication failure remains an invalid probe aggregate."""
+
+    class BrokenProcess:
+        async def communicate(self):
+            raise RuntimeError("simulated process pipe failure")
+
+    async def fake_create_subprocess_exec(*args, **kwargs):
+        return BrokenProcess()
+
+    monkeypatch.setattr(
+        asyncio,
+        "create_subprocess_exec",
+        fake_create_subprocess_exec,
+    )
+
+    result = await run_fping_count(
+        ["10.0.0.8"],
+        packets_per_round=3,
+        timeout_ms=800,
+        executable="/test/fping",
+    )
+
+    assert result["10.0.0.8"].is_valid is False
+    assert result["10.0.0.8"].failure_reason == (
+        "fping execution failed: RuntimeError"
+    )
+
+
+async def test_pinger_persists_one_summary_for_each_enabled_device(
+    db_session,
+    monkeypatch,
+):
+    """One monitoring round writes exactly one aggregate per enabled device."""
+    first = Device(
+        name="edge-a",
+        ip="10.0.0.8",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=True,
+    )
+    second = Device(
+        name="edge-b",
+        ip="10.0.0.9",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=True,
+    )
+    disabled = Device(
+        name="edge-disabled",
+        ip="10.0.0.10",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=False,
+    )
+    db_session.add_all([first, second, disabled])
+    await db_session.commit()
+
+    async def fake_run_fping_count(
+        ips,
+        packets_per_round,
+        timeout_ms,
+        executable,
+    ):
+        assert set(ips) == {"10.0.0.8", "10.0.0.9"}
+        return {
+            "10.0.0.8": ProbeResult("10.0.0.8", 3, 2, 8.5, True),
+            "10.0.0.9": ProbeResult("10.0.0.9", 3, 3, 9.5, True),
+        }
+
+    monkeypatch.setattr(
+        "app.services.pinger.run_fping_count",
+        fake_run_fping_count,
+    )
+
+    changes = await Pinger(Settings()).run_one_round(db_session)
+    records = list(
+        (
+            await db_session.execute(
+                select(PingRecord).order_by(PingRecord.device_id)
+            )
+        )
+        .scalars()
+        .all()
+    )
+
+    assert changes == []
+    assert len(records) == 2
+    assert [record.device_id for record in records] == [first.id, second.id]
+    assert records[0].sent_count == 3
+    assert records[0].received_count == 2
+    assert records[0].packet_loss_percent == pytest.approx(33.3333333333)
+    assert records[0].average_rtt_ms == 8.5
+    assert records[0].is_valid is True
+
+
+async def test_pinger_persists_invalid_probe_without_changing_device_state(
+    db_session,
+    monkeypatch,
+):
+    """A runner failure remains observable but cannot mark a device offline."""
+    device = Device(
+        name="edge-a",
+        ip="10.0.0.8",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=True,
+    )
+    db_session.add(device)
+    await db_session.commit()
+
+    async def fake_run_fping_count(*args, **kwargs):
+        return {
+            device.ip: ProbeResult(
+                device.ip,
+                3,
+                0,
+                None,
+                False,
+                "fping executable not found",
+            )
+        }
+
+    monkeypatch.setattr(
+        "app.services.pinger.run_fping_count",
+        fake_run_fping_count,
+    )
+
+    changes = await Pinger(Settings()).run_one_round(db_session)
+    record = (
+        await db_session.execute(select(PingRecord).where(PingRecord.device_id == device.id))
+    ).scalar_one()
+
+    assert changes == []
+    assert device.current_status == "online"
+    assert record.is_valid is False
+    assert record.packet_loss_percent is None
+    assert record.failure_reason == "fping executable not found"
+
+
+async def test_pinger_evaluates_recent_valid_summaries_and_returns_change(
+    db_session,
+    monkeypatch,
+):
+    """A newly persisted full-loss summary completes the offline streak."""
+    device = Device(
+        name="edge-a",
+        ip="10.0.0.8",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=True,
+        offline_consecutive_rounds=2,
+    )
+    db_session.add(device)
+    await db_session.flush()
+    db_session.add(
+        PingRecord(
+            device_id=device.id,
+            is_alive=False,
+            response_time_ms=None,
+            round_num=1,
+            sent_count=3,
+            received_count=0,
+            packet_loss_percent=100.0,
+            average_rtt_ms=None,
+            is_valid=True,
+            created_at=datetime.now() - timedelta(seconds=30),
+        )
+    )
+    await db_session.commit()
+
+    async def fake_run_fping_count(*args, **kwargs):
+        return {
+            device.ip: ProbeResult(
+                device.ip,
+                3,
+                0,
+                None,
+                True,
+            )
+        }
+
+    monkeypatch.setattr(
+        "app.services.pinger.run_fping_count",
+        fake_run_fping_count,
+    )
+
+    changes = await Pinger(Settings()).run_one_round(db_session)
+
+    assert len(changes) == 1
+    assert changes[0].device is device
+    assert changes[0].old_status == "online"
+    assert changes[0].new_status == "offline"
+    assert changes[0].event_type == "offline"
+    assert device.current_status == "offline"
+    assert device.last_offline_time is not None

backend/tests/test_models_and_migration.py

@@ -0,0 +1,292 @@
+"""Persistence contracts for reliability monitoring data."""
+
+import importlib
+from pathlib import Path
+
+import pytest
+from sqlalchemy import Column, Index, Integer, Table, inspect, text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+import app.models as models
+from app.config import Settings
+from app.models.device import Base
+
+
+@pytest.fixture
+async def db_session() -> AsyncSession:
+    """Provide an isolated database containing the current model metadata."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            yield session
+    finally:
+        await engine.dispose()
+
+
+async def test_probe_record_stores_packet_summary(db_session: AsyncSession):
+    """A probe aggregate retains loss and latency needed by the state machine."""
+    record = models.PingRecord(
+        device_id=1,
+        is_alive=True,
+        response_time_ms=12.5,
+        round_num=1,
+        sent_count=3,
+        received_count=2,
+        packet_loss_percent=33.33,
+        average_rtt_ms=12.5,
+        is_valid=True,
+    )
+    db_session.add(record)
+    await db_session.commit()
+
+    assert record.packet_loss_percent == 33.33
+    assert record.received_count == 2
+
+
+async def test_outbox_defaults_to_pending(db_session: AsyncSession):
+    """New notifications stay eligible for delivery until the dispatcher claims them."""
+    assert hasattr(models, "NotificationOutbox")
+    assert hasattr(models, "NotificationStatus")
+    item = models.NotificationOutbox(
+        alert_event_id=1,
+        message_content="masked summary",
+    )
+    db_session.add(item)
+    await db_session.commit()
+
+    assert item.status == models.NotificationStatus.pending
+
+
+def test_device_policy_prefers_explicit_device_overrides():
+    """A device-specific threshold supersedes only the configured global default."""
+    device = models.Device(
+        probe_packets_per_round=5,
+        degraded_loss_percent=None,
+    )
+
+    policy = models.DeviceMonitoringPolicy.from_device(device, Settings())
+
+    assert policy.probe_packets_per_round == 5
+    assert policy.degraded_loss_percent == 20.0
+
+
+def test_device_policy_uses_global_default_for_invalid_persisted_override():
+    """An invalid stored override cannot weaken a device's safe probe policy."""
+    device = models.Device(
+        probe_packets_per_round=0,
+        degraded_loss_percent=101.0,
+    )
+
+    policy = models.DeviceMonitoringPolicy.from_device(device, Settings())
+
+    assert policy.probe_packets_per_round == 3
+    assert policy.degraded_loss_percent == 20.0
+
+
+async def test_migration_is_idempotent_and_records_revision(tmp_path: Path):
+    """Rerunning the migration preserves the schema and writes one revision marker."""
+    migration = importlib.import_module(
+        "migrations.versions.20260803_reliability_monitoring"
+    )
+    assert hasattr(migration, "run_reliability_migration")
+
+    database_url = f"sqlite+aiosqlite:///{tmp_path / 'monitoring.db'}"
+    engine = create_async_engine(database_url)
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(migration.run_reliability_migration)
+            await connection.run_sync(Base.metadata.create_all)
+        async with engine.begin() as connection:
+            await connection.run_sync(migration.run_reliability_migration)
+            revision_count = await connection.scalar(
+                text(
+                    "SELECT count(*) FROM schema_migrations "
+                    "WHERE revision = '20260803_reliability_monitoring'"
+                )
+            )
+        assert revision_count == 1
+    finally:
+        await engine.dispose()
+
+
+async def test_migration_preserves_legacy_rows_and_owns_only_its_schema(
+    tmp_path: Path,
+):
+    """The revision upgrades legacy data twice without creating unrelated models."""
+    migration = importlib.import_module(
+        "migrations.versions.20260803_reliability_monitoring"
+    )
+    database_url = f"sqlite+aiosqlite:///{tmp_path / 'legacy-monitoring.db'}"
+    engine = create_async_engine(database_url)
+    future_table = Table(
+        "future_unrelated_model",
+        Base.metadata,
+        Column("id", Integer, primary_key=True),
+    )
+    try:
+        async with engine.begin() as connection:
+            await connection.execute(
+                text(
+                    "CREATE TABLE devices ("
+                    "id INTEGER PRIMARY KEY, name VARCHAR(128) NOT NULL, "
+                    "ip VARCHAR(45) NOT NULL, device_type VARCHAR(16) NOT NULL, "
+                    "location VARCHAR(256), project_name VARCHAR(256), "
+                    "tags VARCHAR(512), ping_interval INTEGER, "
+                    "alert_threshold INTEGER, is_enabled BOOLEAN, "
+                    "current_status VARCHAR(16), consecutive_failures INTEGER, "
+                    "last_ping_time DATETIME, last_online_time DATETIME, "
+                    "last_offline_time DATETIME, created_at DATETIME, "
+                    "updated_at DATETIME)"
+                )
+            )
+            await connection.execute(
+                text(
+                    "CREATE TABLE ping_records ("
+                    "id INTEGER PRIMARY KEY, device_id INTEGER NOT NULL, "
+                    "is_alive BOOLEAN NOT NULL, response_time_ms FLOAT, "
+                    "round_num INTEGER NOT NULL, created_at DATETIME)"
+                )
+            )
+            await connection.execute(
+                text(
+                    "CREATE TABLE alert_events ("
+                    "id INTEGER PRIMARY KEY, device_id INTEGER NOT NULL, "
+                    "alert_type VARCHAR(16) NOT NULL, message VARCHAR(1024), "
+                    "start_at DATETIME NOT NULL, end_at DATETIME, "
+                    "duration_minutes INTEGER, is_resolved BOOLEAN, "
+                    "notification_sent BOOLEAN, acknowledged_at DATETIME, "
+                    "created_at DATETIME)"
+                )
+            )
+            await connection.execute(
+                text(
+                    "INSERT INTO devices (id, name, ip, device_type, current_status) "
+                    "VALUES (1, 'legacy-router', '192.0.2.10', 'router', 'online')"
+                )
+            )
+            await connection.execute(
+                text(
+                    "INSERT INTO ping_records "
+                    "(id, device_id, is_alive, response_time_ms, round_num) "
+                    "VALUES (2, 1, 1, 8.5, 7)"
+                )
+            )
+            await connection.execute(
+                text(
+                    "INSERT INTO alert_events "
+                    "(id, device_id, alert_type, message, start_at, notification_sent) "
+                    "VALUES (3, 1, 'offline', 'legacy event', CURRENT_TIMESTAMP, 1)"
+                )
+            )
+
+            await connection.run_sync(migration.run_reliability_migration)
+            table_names = await connection.run_sync(
+                lambda sync_connection: set(
+                    inspect(sync_connection).get_table_names()
+                )
+            )
+            assert "future_unrelated_model" not in table_names
+            assert "notification_outbox" in table_names
+            Base.metadata.remove(future_table)
+            await connection.run_sync(Base.metadata.create_all)
+
+        async with engine.begin() as connection:
+            await connection.run_sync(migration.run_reliability_migration)
+            legacy_device = await connection.execute(
+                text("SELECT name, ip, current_status FROM devices WHERE id = 1")
+            )
+            legacy_probe = await connection.execute(
+                text(
+                    "SELECT device_id, is_alive, response_time_ms, round_num "
+                    "FROM ping_records WHERE id = 2"
+                )
+            )
+            legacy_alert = await connection.execute(
+                text(
+                    "SELECT device_id, alert_type, message, notification_sent "
+                    "FROM alert_events WHERE id = 3"
+                )
+            )
+            probe_columns, alert_columns, probe_indexes = await connection.run_sync(
+                lambda sync_connection: (
+                    {
+                        column["name"]: column
+                        for column in inspect(sync_connection).get_columns(
+                            "ping_records"
+                        )
+                    },
+                    {
+                        column["name"]
+                        for column in inspect(sync_connection).get_columns(
+                            "alert_events"
+                        )
+                    },
+                    {
+                        index["name"]
+                        for index in inspect(sync_connection).get_indexes(
+                            "ping_records"
+                        )
+                    },
+                )
+            )
+
+        assert legacy_device.one() == ("legacy-router", "192.0.2.10", "online")
+        assert legacy_probe.one() == (1, 1, 8.5, 7)
+        assert legacy_alert.one() == (1, "offline", "legacy event", 1)
+        assert probe_columns["is_valid"]["nullable"] is False
+        assert {
+            "sent_count",
+            "received_count",
+            "packet_loss_percent",
+            "average_rtt_ms",
+            "failure_reason",
+        }.issubset(probe_columns)
+        assert "related_event_id" in alert_columns
+        assert "idx_ping_records_device_created_at" in probe_indexes
+    finally:
+        if future_table in Base.metadata.tables.values():
+            Base.metadata.remove(future_table)
+        await engine.dispose()
+
+
+async def test_migration_does_not_create_a_future_ping_record_index(
+    tmp_path: Path,
+):
+    """Historical migration DDL stays limited to its named composite index."""
+    migration = importlib.import_module(
+        "migrations.versions.20260803_reliability_monitoring"
+    )
+    future_index = Index(
+        "idx_future_ping_record_round_num",
+        models.PingRecord.__table__.c.round_num,
+    )
+    engine = create_async_engine(
+        f"sqlite+aiosqlite:///{tmp_path / 'future-index.db'}"
+    )
+    try:
+        async with engine.begin() as connection:
+            await connection.execute(
+                text(
+                    "CREATE TABLE ping_records ("
+                    "id INTEGER PRIMARY KEY, device_id INTEGER NOT NULL, "
+                    "is_alive BOOLEAN NOT NULL, response_time_ms FLOAT, "
+                    "round_num INTEGER NOT NULL, created_at DATETIME)"
+                )
+            )
+            await connection.run_sync(migration.run_reliability_migration)
+            index_names = await connection.run_sync(
+                lambda sync_connection: {
+                    index["name"]
+                    for index in inspect(sync_connection).get_indexes(
+                        "ping_records"
+                    )
+                }
+            )
+
+        assert "idx_ping_records_device_created_at" in index_names
+        assert "idx_future_ping_record_round_num" not in index_names
+    finally:
+        models.PingRecord.__table__.indexes.remove(future_index)
+        await engine.dispose()

backend/tests/test_notification_dispatcher.py

@@ -0,0 +1,327 @@
+"""Persistent notification dispatch and retry behavior tests."""
+
+import json
+import logging
+from datetime import datetime, timedelta
+
+import httpx
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+from app.config import Settings
+from app.models import (
+    AlertEvent,
+    AlertTypeEnum,
+    Base,
+    NotificationOutbox,
+    NotificationStatus,
+)
+from app.services.notification_dispatcher import (
+    DeliveryResult,
+    MAX_EVENTS_PER_MESSAGE,
+    NotificationDispatcher,
+    WeComClient,
+)
+
+
+class FakeWeComClient:
+    """Return one controlled outcome without any external network access."""
+
+    def __init__(self, *results: DeliveryResult):
+        self._results = list(results)
+        self.contents: list[str] = []
+
+    async def send_text(self, content: str) -> DeliveryResult:
+        self.contents.append(content)
+        return self._results.pop(0)
+
+
+@pytest.fixture
+async def db_session() -> AsyncSession:
+    """Provide an isolated outbox database."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            yield session
+    finally:
+        await engine.dispose()
+
+
+async def test_retryable_failure_reschedules(
+    db_session: AsyncSession,
+):
+    """A transient failure must remain pending with exponential retry evidence."""
+    frozen_time = datetime(2026, 8, 4, 10, 0)
+    pending_message = NotificationOutbox(
+        alert_event_id=1,
+        message_content="设备离线",
+    )
+    db_session.add(pending_message)
+    await db_session.commit()
+    client = FakeWeComClient(DeliveryResult(False, "timeout", True))
+
+    await NotificationDispatcher(client).dispatch_due(
+        db_session,
+        frozen_time,
+    )
+
+    assert pending_message.status == NotificationStatus.pending
+    assert pending_message.attempt_count == 1
+    assert pending_message.next_attempt_at == datetime(2026, 8, 4, 10, 0, 30)
+
+
+async def test_success_marks_outbox_and_event_delivered(
+    db_session: AsyncSession,
+):
+    """A successful transport result updates both delivery evidence records."""
+    now = datetime(2026, 8, 4, 10, 0)
+    event = AlertEvent(
+        device_id=7,
+        alert_type=AlertTypeEnum.offline,
+        message="full loss",
+        start_at=now,
+        is_resolved=False,
+    )
+    db_session.add(event)
+    await db_session.flush()
+    message = NotificationOutbox(
+        alert_event_id=event.id,
+        message_content="设备 7 离线",
+    )
+    db_session.add(message)
+    await db_session.commit()
+
+    summary = await NotificationDispatcher(
+        FakeWeComClient(DeliveryResult(True)),
+        _enabled_settings(),
+    ).dispatch_due(db_session, now)
+
+    assert summary.sent == 1
+    assert message.status == NotificationStatus.sent
+    assert message.sent_at == now
+    assert message.attempt_count == 1
+    assert event.notification_sent is True
+    assert event.notification_attempts == 1
+    assert event.last_notification_error is None
+
+
+async def test_non_retryable_or_exhausted_failure_is_terminal(
+    db_session: AsyncSession,
+):
+    """The configured attempt ceiling prevents an infinite retry loop."""
+    now = datetime(2026, 8, 4, 10, 0)
+    message = NotificationOutbox(
+        alert_event_id=99,
+        message_content="bad recipient",
+        attempt_count=1,
+    )
+    db_session.add(message)
+    await db_session.commit()
+    runtime_settings = _enabled_settings(wecom_notification_max_attempts=2)
+
+    summary = await NotificationDispatcher(
+        FakeWeComClient(DeliveryResult(False, "HTTP 503", True)),
+        runtime_settings,
+    ).dispatch_due(db_session, now)
+
+    assert summary.failed == 1
+    assert message.status == NotificationStatus.failed
+    assert message.attempt_count == 2
+    assert message.next_attempt_at is None
+
+
+async def test_only_due_messages_are_dispatched_in_fixed_safe_batches(
+    db_session: AsyncSession,
+):
+    """Large bursts are split while future retries remain untouched."""
+    now = datetime(2026, 8, 4, 10, 0)
+    due = [
+        NotificationOutbox(alert_event_id=index, message_content=f"事件 {index}")
+        for index in range(MAX_EVENTS_PER_MESSAGE + 1)
+    ]
+    future = NotificationOutbox(
+        alert_event_id=100,
+        message_content="未来重试",
+        next_attempt_at=now + timedelta(minutes=1),
+    )
+    db_session.add_all([*due, future])
+    await db_session.commit()
+    client = FakeWeComClient(DeliveryResult(True), DeliveryResult(True))
+
+    summary = await NotificationDispatcher(
+        client,
+        _enabled_settings(),
+    ).dispatch_due(db_session, now)
+
+    assert summary.sent == MAX_EVENTS_PER_MESSAGE + 1
+    assert len(client.contents) == 2
+    assert future.status == NotificationStatus.pending
+    assert future.attempt_count == 0
+
+
+def _enabled_settings(**overrides) -> Settings:
+    values = {
+        "wecom_notification_enabled": True,
+        "WECOM_CORP_ID": "test-corp",
+        "WECOM_AGENT_ID": 1000001,
+        "WECOM_APP_SECRET": "test-secret",
+    }
+    values.update(overrides)
+    return Settings(**values)
+
+
+@pytest.mark.parametrize(
+    ("to_party", "expected_recipient", "unexpected_key"),
+    [("", ("touser", "@all"), "toparty"), ("2|3", ("toparty", "2|3"), "touser")],
+)
+async def test_wecom_uses_application_scope_unless_department_is_explicit(
+    to_party,
+    expected_recipient,
+    unexpected_key,
+):
+    """Recipient payloads preserve the app visibility boundary by default."""
+    payloads: list[dict] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/gettoken"):
+            return httpx.Response(
+                200,
+                json={"errcode": 0, "access_token": "token-a", "expires_in": 7200},
+            )
+        payloads.append(json.loads(request.content))
+        return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
+        client = WeComClient(
+            _enabled_settings(WECOM_TO_PARTY=to_party),
+            http_client=http,
+        )
+        result = await client.send_text("controlled test message")
+
+    key, value = expected_recipient
+    assert result.success is True
+    assert payloads[0][key] == value
+    assert unexpected_key not in payloads[0]
+    assert payloads[0]["agentid"] == 1000001
+
+
+async def test_wecom_caches_token_until_expiry_margin():
+    """Repeated messages do not request a new token before its safe expiry."""
+    calls = {"token": 0, "message": 0}
+    current_time = [datetime(2026, 8, 4, 10, 0)]
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/gettoken"):
+            calls["token"] += 1
+            return httpx.Response(
+                200,
+                json={"errcode": 0, "access_token": "token-a", "expires_in": 120},
+            )
+        calls["message"] += 1
+        return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
+        client = WeComClient(
+            _enabled_settings(),
+            http_client=http,
+            now=lambda: current_time[0],
+        )
+        await client.send_text("first")
+        current_time[0] += timedelta(seconds=30)
+        await client.send_text("second")
+
+    assert calls == {"token": 1, "message": 2}
+
+
+async def test_wecom_refreshes_token_after_safety_margin():
+    """A token inside the safety margin cannot be reused for a new message."""
+    token_calls = 0
+    current_time = [datetime(2026, 8, 4, 10, 0)]
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal token_calls
+        if request.url.path.endswith("/gettoken"):
+            token_calls += 1
+            return httpx.Response(
+                200,
+                json={
+                    "errcode": 0,
+                    "access_token": f"token-{token_calls}",
+                    "expires_in": 120,
+                },
+            )
+        return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
+        client = WeComClient(
+            _enabled_settings(),
+            http_client=http,
+            now=lambda: current_time[0],
+        )
+        await client.send_text("first")
+        current_time[0] += timedelta(seconds=61)
+        await client.send_text("second")
+
+    assert token_calls == 2
+
+
+async def test_wecom_rejects_non_object_json_without_raising():
+    """Malformed external response shapes return sanitized retry evidence."""
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=["unexpected", "shape"])
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
+        result = await WeComClient(
+            _enabled_settings(),
+            http_client=http,
+        ).send_text("controlled")
+
+    assert result == DeliveryResult(False, "invalid token response", True)
+
+
+@pytest.mark.parametrize("status_code", [429, 500, 503])
+async def test_wecom_marks_rate_limit_and_server_errors_retryable(status_code):
+    """Transient HTTP classes reach the outbox retry path without response leakage."""
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/gettoken"):
+            return httpx.Response(
+                200,
+                json={"errcode": 0, "access_token": "token-a", "expires_in": 7200},
+            )
+        return httpx.Response(status_code, text="sensitive-response-body")
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
+        result = await WeComClient(
+            _enabled_settings(),
+            http_client=http,
+        ).send_text("sensitive-message-content")
+
+    assert result.success is False
+    assert result.retryable is True
+    assert result.error == f"HTTP {status_code}"
+
+
+async def test_wecom_transport_failure_is_sanitized_and_retryable(caplog):
+    """Logs and returned evidence never expose secrets, tokens, bodies, or content."""
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/gettoken"):
+            return httpx.Response(
+                200,
+                json={"errcode": 0, "access_token": "token-a", "expires_in": 7200},
+            )
+        raise httpx.ConnectError("response-sensitive", request=request)
+
+    caplog.set_level(logging.WARNING)
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
+        result = await WeComClient(
+            _enabled_settings(WECOM_APP_SECRET="secret-sensitive"),
+            http_client=http,
+        ).send_text("message-sensitive")
+
+    assert result == DeliveryResult(False, "transport error", True)
+    assert "secret-sensitive" not in caplog.text
+    assert "token-a" not in caplog.text
+    assert "response-sensitive" not in caplog.text
+    assert "message-sensitive" not in caplog.text

backend/tests/test_settings_validation.py

@@ -0,0 +1,68 @@
+import pytest
+from pydantic import ValidationError
+
+from app.config import Settings
+from app.services.settings_validation import validate_runtime_settings
+
+
+def test_production_rejects_legacy_jwt_secret():
+    settings = Settings(
+        environment="production",
+        secret_key="change-me-to-a-long-random-string",
+    )
+
+    with pytest.raises(ValueError, match="SECRET_KEY"):
+        validate_runtime_settings(settings)
+
+
+@pytest.mark.parametrize("secret_key", ["", "   ", "\t\n"])
+def test_production_rejects_blank_jwt_secret(secret_key):
+    settings = Settings(environment="production", secret_key=secret_key)
+
+    with pytest.raises(ValueError, match="SECRET_KEY"):
+        validate_runtime_settings(settings)
+
+
+@pytest.mark.parametrize("environment", ["prod", "Production", "staging"])
+def test_environment_must_be_an_explicit_supported_value(environment):
+    with pytest.raises(ValidationError, match="environment"):
+        Settings(environment=environment)
+
+
+def test_probe_packet_count_must_be_positive():
+    with pytest.raises(ValidationError):
+        Settings(probe_packets_per_round=0)
+
+
+@pytest.mark.parametrize(
+    ("field", "invalid_value"),
+    [
+        ("offline_consecutive_rounds", 0),
+        ("degraded_window_rounds", 1),
+        ("degraded_loss_percent", 0.0),
+        ("recovery_consecutive_clean_rounds", 0),
+        ("wecom_notification_max_attempts", 0),
+        ("wecom_retry_base_seconds", 0),
+    ],
+)
+def test_monitoring_and_retry_settings_reject_values_below_their_bounds(field, invalid_value):
+    with pytest.raises(ValidationError):
+        Settings(**{field: invalid_value})
+
+
+def test_production_rejects_tls_disabled_casdoor():
+    settings = Settings(
+        environment="production",
+        secret_key="a-safe-production-secret",
+        casdoor_endpoint="http://casdoor.internal",
+    )
+
+    with pytest.raises(ValueError, match="CASDOOR"):
+        validate_runtime_settings(settings)
+
+
+def test_enabled_wecom_delivery_requires_all_credentials():
+    settings = Settings(wecom_notification_enabled=True, WECOM_CORP_ID="corp")
+
+    with pytest.raises(ValueError, match="WECOM"):
+        validate_runtime_settings(settings)

backend/tests/test_state_machine.py

@@ -0,0 +1,139 @@
+"""Unit tests for the connectivity health state machine."""
+
+from dataclasses import dataclass
+
+import pytest
+
+from app.models.device import DeviceMonitoringPolicy
+from app.services.state_machine import evaluate_health
+
+
+@dataclass(frozen=True)
+class RecentProbe:
+    """Minimal real-data-shaped record consumed by the pure state machine."""
+
+    sent_count: int
+    received_count: int
+    is_valid: bool = True
+
+
+def loss(sent_count: int) -> RecentProbe:
+    return RecentProbe(sent_count=sent_count, received_count=0)
+
+
+def partial(sent_count: int, received_count: int) -> RecentProbe:
+    return RecentProbe(
+        sent_count=sent_count,
+        received_count=received_count,
+    )
+
+
+def clean(sent_count: int) -> RecentProbe:
+    return RecentProbe(sent_count=sent_count, received_count=sent_count)
+
+
+@pytest.fixture
+def policy() -> DeviceMonitoringPolicy:
+    return DeviceMonitoringPolicy(
+        probe_packets_per_round=3,
+        offline_consecutive_rounds=2,
+        degraded_window_rounds=5,
+        degraded_loss_percent=20.0,
+        recovery_consecutive_clean_rounds=3,
+    )
+
+
+def test_two_full_loss_rounds_take_device_offline(policy):
+    """Configured consecutive full-loss rounds transition to offline once."""
+    decision = evaluate_health("online", [loss(3), loss(3)], policy)
+
+    assert (decision.next_status, decision.event_type) == (
+        "offline",
+        "offline",
+    )
+    assert decision.should_notify is True
+
+
+def test_offline_priority_wins_over_aggregate_degraded_loss(policy):
+    """Full-loss streaks are classified offline even when the window also degrades."""
+    recent = [partial(3, 2)] * 3 + [loss(3), loss(3)]
+
+    decision = evaluate_health("online", recent, policy)
+
+    assert decision.next_status == "offline"
+    assert decision.event_type == "offline"
+
+
+def test_five_round_window_with_loss_is_degraded(policy):
+    """Aggregate loss at the configured window threshold is degraded."""
+    decision = evaluate_health(
+        "online",
+        [partial(3, 2)] * 5,
+        policy,
+    )
+
+    assert decision.next_status == "degraded"
+    assert decision.event_type == "degraded"
+
+
+def test_degraded_requires_a_complete_window(policy):
+    """A partial history cannot satisfy an aggregate-window decision."""
+    decision = evaluate_health(
+        "online",
+        [partial(3, 2)] * 4,
+        policy,
+    )
+
+    assert decision.next_status == "online"
+    assert decision.event_type is None
+
+
+def test_three_clean_rounds_recovers(policy):
+    """An alerted device recovers only after the configured clean streak."""
+    decision = evaluate_health(
+        "offline",
+        [clean(3)] * 3,
+        policy,
+    )
+
+    assert (decision.next_status, decision.event_type) == (
+        "online",
+        "recovered",
+    )
+    assert decision.should_notify is True
+
+
+def test_clean_online_device_does_not_emit_duplicate_event(policy):
+    """Clean probes retain an already-online state without notifying."""
+    decision = evaluate_health(
+        "online",
+        [clean(3)] * 5,
+        policy,
+    )
+
+    assert decision.next_status == "online"
+    assert decision.event_type is None
+    assert decision.should_notify is False
+
+
+def test_latest_invalid_probe_retains_state(policy):
+    """A malformed current result cannot reuse old loss to trigger a fault."""
+    recent = [loss(3), RecentProbe(3, 0, is_valid=False)]
+
+    decision = evaluate_health("online", recent, policy)
+
+    assert decision.next_status == "online"
+    assert decision.event_type is None
+    assert decision.should_notify is False
+
+
+def test_offline_does_not_recover_before_clean_streak_is_complete(policy):
+    """Recovery debounce retains offline until enough clean rounds exist."""
+    decision = evaluate_health(
+        "offline",
+        [clean(3), clean(3)],
+        policy,
+    )
+
+    assert decision.next_status == "offline"
+    assert decision.event_type is None

docker-compose.yml

@@ -6,10 +6,9 @@ services:
     env_file:
       - ./backend/.env
     environment:
-      - DATABASE_URL=postgresql+asyncpg://pingwatch:pingwatch123@db:5432/pingwatch
       - TZ=Asia/Shanghai
     ports:
-      - "8001:8000"
+      - "127.0.0.1:8001:8000"
     depends_on:
       db:
         condition: service_healthy
@@ -18,8 +17,7 @@ services:
     networks:
       - pingwatch-net
     cap_add:
-      - NET_RAW          # 允许 ICMP ping
+      - NET_RAW
-      - NET_ADMIN        # 允许原始套接字
 
   frontend:
     build: ./frontend
@@ -40,9 +38,9 @@ services:
     restart: unless-stopped
     environment:
       - TZ=Asia/Shanghai
-      - POSTGRES_DB=pingwatch
+      - POSTGRES_DB=${POSTGRES_DB:-pingwatch}
-      - POSTGRES_USER=pingwatch
+      - POSTGRES_USER=${POSTGRES_USER:-pingwatch}
-      - POSTGRES_PASSWORD=pingwatch123
+      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD:?POSTGRES_PASSWORD must be set in backend/.env}
     volumes:
       - postgres_data:/var/lib/postgresql/data
     healthcheck: