feat(channel/monitoring): 新增完整的频道监控模块

实现了包括事件循环监控、健康检查、状态聚合、诊断心跳、稳定性追踪在内的全套监控能力，提供指标采集、就绪检查、告警推送等功能
2026-05-21 10:27:34 +08:00 · 2026-05-21 10:27:34 +08:00 · 79c6e6b74f
commit 79c6e6b74f
parent 0b19470c70
10 changed files with 2563 additions and 0 deletions
--- a/backend/package/yuxi/channel/monitoring/init.py
+++ b/backend/package/yuxi/channel/monitoring/init.py
@ -0,0 +1,123 @@
+from yuxi.channel.monitoring.diagnostic_heartbeat import (
+    DiagnosticHeartbeat,
+    HeartbeatReport,
+    MemoryPressureEvent,
+    MemorySnapshot,
+    diagnostic_heartbeat,
+)
+from yuxi.channel.monitoring.diagnostic_phase import (
+    PhaseSnapshot,
+    diagnostic_phase,
+    get_current_phase,
+    get_recent_phases,
+    record_diagnostic_phase,
+    reset_diagnostic_phases_for_test,
+)
+from yuxi.channel.monitoring.diagnostic_stability import (
+    StabilityEvent,
+    StabilityRingBuffer,
+    get_stability_snapshot,
+    install_stability_crash_hook,
+    record_stability_event,
+    write_stability_bundle,
+)
+from yuxi.channel.monitoring.event_loop_monitor import (
+    EventLoopHealth,
+    EventLoopMonitor,
+    EventLoopStats,
+    event_loop_monitor,
+)
+from yuxi.channel.monitoring.event_loop_ready import (
+    EventLoopReadyResult,
+    wait_for_event_loop_ready,
+)
+from yuxi.channel.monitoring.health_monitor import (
+    ChannelHealthEvaluation,
+    ChannelHealthMetrics,
+    ChannelHealthMonitor,
+    channel_health_monitor,
+    evaluate_channel_health,
+    resolve_restart_reason,
+)
+from yuxi.channel.monitoring.metrics import (
+    metrics_registry,
+    send_alert_webhook,
+    active_sessions,
+    active_tasks,
+    channel_health,
+    event_loop_delay_p99,
+    event_loop_utilization,
+    memory_gc_objects,
+    memory_rss_mb,
+    stuck_sessions,
+)
+from yuxi.channel.monitoring.readiness import (
+    ReadinessChecker,
+    ReadinessResult,
+    readiness_checker,
+)
+from yuxi.channel.monitoring.state_aggregator import (
+    ChannelHealthState,
+    ConfigHealthState,
+    DiagnosticState,
+    GatewayConnectionState,
+    MonitoringState,
+    ReadinessState,
+    StateAggregator,
+    UnifiedState,
+    state_aggregator,
+)
+
+__all__ = [
+    "ChannelHealthEvaluation",
+    "ChannelHealthMetrics",
+    "ChannelHealthMonitor",
+    "ChannelHealthState",
+    "ConfigHealthState",
+    "DiagnosticHeartbeat",
+    "DiagnosticState",
+    "EventLoopHealth",
+    "EventLoopMonitor",
+    "EventLoopReadyResult",
+    "EventLoopStats",
+    "GatewayConnectionState",
+    "HeartbeatReport",
+    "MemoryPressureEvent",
+    "MemorySnapshot",
+    "MonitoringState",
+    "PhaseSnapshot",
+    "ReadinessChecker",
+    "ReadinessResult",
+    "ReadinessState",
+    "StabilityEvent",
+    "StabilityRingBuffer",
+    "StateAggregator",
+    "UnifiedState",
+    "channel_health_monitor",
+    "diagnostic_heartbeat",
+    "diagnostic_phase",
+    "event_loop_monitor",
+    "evaluate_channel_health",
+    "get_current_phase",
+    "get_recent_phases",
+    "get_stability_snapshot",
+    "install_stability_crash_hook",
+    "readiness_checker",
+    "record_diagnostic_phase",
+    "record_stability_event",
+    "reset_diagnostic_phases_for_test",
+    "resolve_restart_reason",
+    "state_aggregator",
+    "wait_for_event_loop_ready",
+    "write_stability_bundle",
+    "metrics_registry",
+    "send_alert_webhook",
+    "active_sessions",
+    "active_tasks",
+    "channel_health",
+    "event_loop_delay_p99",
+    "event_loop_utilization",
+    "memory_gc_objects",
+    "memory_rss_mb",
+    "stuck_sessions",
+]
--- a/backend/package/yuxi/channel/monitoring/diagnostic_heartbeat.py
+++ b/backend/package/yuxi/channel/monitoring/diagnostic_heartbeat.py
@ -0,0 +1,473 @@
+import asyncio
+import gc
+import logging
+import time
+from dataclasses import dataclass, field
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_HEARTBEAT_INTERVAL = 30
+STUCK_SESSION_TIMEOUT = 600
+MAX_CONSECUTIVE_FAILURES = 5
+
+MB = 1024 * 1024
+DEFAULT_RSS_WARNING_MB = 1536
+DEFAULT_RSS_CRITICAL_MB = 3072
+DEFAULT_DATA_WARNING_MB = 1024
+DEFAULT_DATA_CRITICAL_MB = 2048
+DEFAULT_GC_OBJECTS_WARNING = 500000
+DEFAULT_GC_SAMPLE_INTERVAL = 5
+DEFAULT_RSS_GROWTH_WARNING_MB = 512
+DEFAULT_RSS_GROWTH_CRITICAL_MB = 1024
+DEFAULT_GROWTH_WINDOW_S = 600
+DEFAULT_PRESSURE_COOLDOWN_S = 300
+
+
+@dataclass
+class MemorySnapshot:
+    rss_mb: float = 0.0
+    vms_mb: float = 0.0
+    data_mb: float = 0.0
+    gc_objects: int = 0
+    gc_g0_count: int = 0
+    gc_g1_count: int = 0
+    gc_g2_count: int = 0
+    timestamp: float = 0.0
+
+
+@dataclass
+class MemoryPressureEvent:
+    level: str  # "warning" | "critical"
+    reason: str  # "rss_threshold" | "data_threshold" | "rss_growth"
+    rss_mb: float = 0.0
+    threshold_mb: float = 0.0
+    rss_growth_mb: float = 0.0
+    window_s: float = 0.0
+    timestamp: float = 0.0
+
+
+@dataclass
+class HeartbeatReport:
+    active_sessions: int = 0
+    stuck_sessions: int = 0
+    stuck_details: list[dict] = field(default_factory=list)
+    tool_loop_sessions: int = 0
+    tool_loop_details: list[dict] = field(default_factory=list)
+    memory: MemorySnapshot = field(default_factory=MemorySnapshot)
+    active_tasks: int = 0
+    uptime_seconds: float = 0.0
+    process_start_timestamp: float = 0.0
+    timestamp: float = 0.0
+    warnings: list[str] = field(default_factory=list)
+    memory_pressure: MemoryPressureEvent | None = None
+
+    def to_dict(self) -> dict:
+        result = {
+            "active_sessions": self.active_sessions,
+            "stuck_sessions": self.stuck_sessions,
+            "stuck_details": self.stuck_details,
+            "tool_loop_sessions": self.tool_loop_sessions,
+            "tool_loop_details": self.tool_loop_details,
+            "memory": {
+                "rss_mb": self.memory.rss_mb,
+                "vms_mb": self.memory.vms_mb,
+                "data_mb": self.memory.data_mb,
+                "gc_objects": self.memory.gc_objects,
+                "gc_g0_count": self.memory.gc_g0_count,
+                "gc_g1_count": self.memory.gc_g1_count,
+                "gc_g2_count": self.memory.gc_g2_count,
+                "timestamp": self.memory.timestamp,
+            },
+            "active_tasks": self.active_tasks,
+            "uptime_seconds": self.uptime_seconds,
+            "process_start_timestamp": self.process_start_timestamp,
+            "timestamp": self.timestamp,
+            "warnings": self.warnings,
+        }
+        if self.memory_pressure:
+            result["memory_pressure"] = {
+                "level": self.memory_pressure.level,
+                "reason": self.memory_pressure.reason,
+                "rss_mb": self.memory_pressure.rss_mb,
+                "threshold_mb": self.memory_pressure.threshold_mb,
+                "rss_growth_mb": self.memory_pressure.rss_growth_mb,
+                "window_s": self.memory_pressure.window_s,
+                "timestamp": self.memory_pressure.timestamp,
+            }
+        return result
+
+
+class DiagnosticHeartbeat:
+    """诊断心跳：30s 间隔检测卡住 session、内存快照 + 内存压力。
+
+    集成到 FastAPI lifespan 后启动，可被 health endpoint 消费。
+    """
+
+    def __init__(
+        self,
+        interval: int = DEFAULT_HEARTBEAT_INTERVAL,
+        stuck_timeout: int = STUCK_SESSION_TIMEOUT,
+        rss_warning_mb: float = DEFAULT_RSS_WARNING_MB,
+        rss_critical_mb: float = DEFAULT_RSS_CRITICAL_MB,
+        data_warning_mb: float = DEFAULT_DATA_WARNING_MB,
+        data_critical_mb: float = DEFAULT_DATA_CRITICAL_MB,
+        rss_growth_warning_mb: float = DEFAULT_RSS_GROWTH_WARNING_MB,
+        rss_growth_critical_mb: float = DEFAULT_RSS_GROWTH_CRITICAL_MB,
+        growth_window_s: float = DEFAULT_GROWTH_WINDOW_S,
+        pressure_cooldown_s: float = DEFAULT_PRESSURE_COOLDOWN_S,
+        gc_objects_warning: int = DEFAULT_GC_OBJECTS_WARNING,
+        gc_sample_interval: int = DEFAULT_GC_SAMPLE_INTERVAL,
+    ):
+        self._interval = interval
+        self._stuck_timeout = stuck_timeout
+        self._rss_warning_mb = rss_warning_mb
+        self._rss_critical_mb = rss_critical_mb
+        self._data_warning_mb = data_warning_mb
+        self._data_critical_mb = data_critical_mb
+        self._rss_growth_warning_mb = rss_growth_warning_mb
+        self._rss_growth_critical_mb = rss_growth_critical_mb
+        self._growth_window_s = growth_window_s
+        self._pressure_cooldown_s = pressure_cooldown_s
+        self._gc_objects_warning = gc_objects_warning
+        self._gc_sample_interval = gc_sample_interval
+        self._running = False
+        self._task: asyncio.Task | None = None
+        self._last_report: HeartbeatReport | None = None
+        self._last_memory: MemorySnapshot | None = None
+        self._last_pressure_at: dict[str, float] = {}
+        self._start_time: float | None = None
+        self._start_monotonic: float | None = None
+        self._consecutive_failures = 0
+        self._cycle_count = 0
+
+    @property
+    def last_report(self) -> HeartbeatReport | None:
+        return self._last_report
+
+    @staticmethod
+    def _resolve_start_time() -> float:
+        try:
+            import psutil
+
+            return psutil.Process().create_time()
+        except Exception:
+            return time.time()
+
+    async def start(self):
+        self._running = True
+        self._start_time = self._resolve_start_time()
+        self._start_monotonic = time.monotonic()
+        self._task = asyncio.create_task(self._loop())
+        logger.info(
+            "DiagnosticHeartbeat started (interval=%ds, stuck_timeout=%ds)",
+            self._interval,
+            self._stuck_timeout,
+        )
+
+    async def stop(self):
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None
+        logger.info("DiagnosticHeartbeat stopped")
+
+    async def _loop(self):
+        while self._running:
+            await asyncio.sleep(self._interval)
+            try:
+                self._last_report = await self._collect()
+                self._consecutive_failures = 0
+                await self._push_metrics(self._last_report)
+                if self._last_report.warnings:
+                    for w in self._last_report.warnings:
+                        logger.warning("DiagnosticHeartbeat: %s", w)
+                if self._last_report.memory_pressure:
+                    mp = self._last_report.memory_pressure
+                    logger.warning(
+                        "DiagnosticHeartbeat: 内存压力 %s (reason=%s, rss=%.0fMB, threshold=%.0fMB)",
+                        mp.level,
+                        mp.reason,
+                        mp.rss_mb,
+                        mp.threshold_mb,
+                    )
+            except Exception:
+                self._consecutive_failures += 1
+                logger.exception(
+                    "DiagnosticHeartbeat: collect failed (consecutive=%d/%d)",
+                    self._consecutive_failures,
+                    MAX_CONSECUTIVE_FAILURES,
+                )
+                if self._consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
+                    logger.critical(
+                        "DiagnosticHeartbeat: %d consecutive collect failures, heartbeat may be broken",
+                        self._consecutive_failures,
+                    )
+
+    async def _collect(self) -> HeartbeatReport:
+        now = time.monotonic()
+        self._cycle_count += 1
+        memory = await self._capture_memory(now)
+        self._last_memory = memory
+
+        snapshots = _get_all_snapshots()
+        stuck_details, stuck_count = await self._detect_stuck_sessions(now, snapshots)
+        active_tasks = len(asyncio.all_tasks())
+        active_sessions = _count_active_sessions_from(snapshots)
+
+        uptime_seconds = now - self._start_monotonic if self._start_monotonic else 0.0
+        process_start_timestamp = self._start_time or 0.0
+
+        memory_pressure = self._detect_memory_pressure(memory, now)
+
+        warnings: list[str] = []
+        if stuck_count > 0:
+            warnings.append(f"检测到 {stuck_count} 个卡住 session")
+        if memory.rss_mb > self._rss_warning_mb:
+            warnings.append(f"内存 RSS {memory.rss_mb:.0f}MB > {self._rss_warning_mb:.0f}MB")
+        if memory.data_mb > self._data_warning_mb:
+            warnings.append(f"内存 DATA {memory.data_mb:.0f}MB > {self._data_warning_mb:.0f}MB")
+        if memory.gc_objects > self._gc_objects_warning:
+            warnings.append(f"GC 对象数 {memory.gc_objects} > {self._gc_objects_warning}")
+        if memory_pressure:
+            level_label = "严重" if memory_pressure.level == "critical" else ""
+            warnings.append(f"内存压力{level_label}: {memory_pressure.reason} (RSS={memory_pressure.rss_mb:.0f}MB)")
+
+        return HeartbeatReport(
+            active_sessions=active_sessions,
+            stuck_sessions=stuck_count,
+            stuck_details=stuck_details,
+            tool_loop_sessions=0,
+            tool_loop_details=[],
+            memory=memory,
+            active_tasks=active_tasks,
+            uptime_seconds=uptime_seconds,
+            process_start_timestamp=process_start_timestamp,
+            timestamp=now,
+            warnings=warnings,
+            memory_pressure=memory_pressure,
+        )
+
+    async def _push_metrics(self, report: HeartbeatReport):
+        try:
+            from yuxi.channel.monitoring.metrics import (
+                active_sessions,
+                active_tasks,
+                memory_gc_objects,
+                memory_rss_mb,
+                stuck_sessions,
+            )
+
+            active_sessions.set(report.active_sessions)
+            stuck_sessions.set(report.stuck_sessions)
+            memory_rss_mb.set(report.memory.rss_mb)
+            memory_gc_objects.set(report.memory.gc_objects)
+            active_tasks.set(report.active_tasks)
+        except Exception:
+            pass
+
+    def _detect_memory_pressure(self, current: MemorySnapshot, now: float) -> MemoryPressureEvent | None:
+        pressure = _check_threshold_pressure(
+            current,
+            rss_warning_mb=self._rss_warning_mb,
+            rss_critical_mb=self._rss_critical_mb,
+        )
+        if pressure is None:
+            pressure = _check_data_threshold_pressure(
+                current,
+                data_warning_mb=self._data_warning_mb,
+                data_critical_mb=self._data_critical_mb,
+            )
+        if pressure is None:
+            pressure = _check_growth_pressure(
+                self._last_memory,
+                current,
+                now,
+                growth_window_s=self._growth_window_s,
+                rss_growth_warning_mb=self._rss_growth_warning_mb,
+                rss_growth_critical_mb=self._rss_growth_critical_mb,
+            )
+
+        if pressure is None:
+            return None
+
+        key = pressure.reason
+        last_at = self._last_pressure_at.get(key, 0)
+        if now - last_at < self._pressure_cooldown_s:
+            return None
+
+        self._last_pressure_at[key] = now
+        return pressure
+
+    async def _detect_stuck_sessions(self, now: float, snapshots: dict) -> tuple[list[dict], int]:
+        stuck: list[dict] = []
+        for key, snap in snapshots.items():
+            if snap.last_event_at and (now - snap.last_event_at) > self._stuck_timeout:
+                state_value = _get_state_value(snap.state)
+                if snap.connected and state_value in ("running", "retrying"):
+                    stuck.append(
+                        {
+                            "key": key,
+                            "state": state_value,
+                            "last_event_at": snap.last_event_at,
+                            "idle_seconds": round(now - snap.last_event_at, 1),
+                        }
+                    )
+        return stuck, len(stuck)
+
+    async def _capture_memory(self, now: float) -> MemorySnapshot:
+        snapshot = MemorySnapshot(timestamp=now)
+        try:
+            import psutil
+
+            proc = psutil.Process()
+            mem = proc.memory_info()
+            snapshot.rss_mb = round(mem.rss / 1024 / 1024, 2)
+            snapshot.vms_mb = round(mem.vms / 1024 / 1024, 2)
+            snapshot.data_mb = round(getattr(mem, "data", 0) / 1024 / 1024, 2)
+        except Exception:
+            pass
+
+        gc_counts = gc.get_count()
+        snapshot.gc_g0_count = gc_counts[0]
+        snapshot.gc_g1_count = gc_counts[1]
+        snapshot.gc_g2_count = gc_counts[2]
+
+        if self._cycle_count % self._gc_sample_interval == 0:
+            snapshot.gc_objects = len(gc.get_objects())
+
+        return snapshot
+
+    def get_memory_trend(self) -> dict:
+        if self._last_memory:
+            return {
+                "rss_mb": self._last_memory.rss_mb,
+                "vms_mb": self._last_memory.vms_mb,
+                "data_mb": self._last_memory.data_mb,
+                "gc_objects": self._last_memory.gc_objects,
+                "timestamp": self._last_memory.timestamp,
+            }
+        return {}
+
+    def reset(self):
+        self._last_memory = None
+        self._last_pressure_at.clear()
+        self._last_report = None
+        self._cycle_count = 0
+
+
+def _check_threshold_pressure(
+    memory: MemorySnapshot,
+    *,
+    rss_warning_mb: float,
+    rss_critical_mb: float,
+) -> MemoryPressureEvent | None:
+    if memory.rss_mb >= rss_critical_mb:
+        return MemoryPressureEvent(
+            level="critical",
+            reason="rss_threshold",
+            rss_mb=memory.rss_mb,
+            threshold_mb=rss_critical_mb,
+            timestamp=memory.timestamp,
+        )
+    if memory.rss_mb >= rss_warning_mb:
+        return MemoryPressureEvent(
+            level="warning",
+            reason="rss_threshold",
+            rss_mb=memory.rss_mb,
+            threshold_mb=rss_warning_mb,
+            timestamp=memory.timestamp,
+        )
+    return None
+
+
+def _check_data_threshold_pressure(
+    memory: MemorySnapshot,
+    *,
+    data_warning_mb: float,
+    data_critical_mb: float,
+) -> MemoryPressureEvent | None:
+    if memory.data_mb >= data_critical_mb:
+        return MemoryPressureEvent(
+            level="critical",
+            reason="data_threshold",
+            rss_mb=memory.rss_mb,
+            threshold_mb=data_critical_mb,
+            timestamp=memory.timestamp,
+        )
+    if memory.data_mb >= data_warning_mb:
+        return MemoryPressureEvent(
+            level="warning",
+            reason="data_threshold",
+            rss_mb=memory.rss_mb,
+            threshold_mb=data_warning_mb,
+            timestamp=memory.timestamp,
+        )
+    return None
+
+
+def _check_growth_pressure(
+    previous: MemorySnapshot | None,
+    current: MemorySnapshot,
+    now: float,
+    *,
+    growth_window_s: float,
+    rss_growth_warning_mb: float,
+    rss_growth_critical_mb: float,
+) -> MemoryPressureEvent | None:
+    if previous is None or previous.rss_mb <= 0:
+        return None
+
+    window_s = now - previous.timestamp
+    if window_s <= 0 or window_s > growth_window_s:
+        return None
+
+    rss_growth_mb = current.rss_mb - previous.rss_mb
+    if rss_growth_mb <= 0:
+        return None
+
+    if rss_growth_mb >= rss_growth_critical_mb:
+        return MemoryPressureEvent(
+            level="critical",
+            reason="rss_growth",
+            rss_mb=current.rss_mb,
+            threshold_mb=rss_growth_critical_mb,
+            rss_growth_mb=round(rss_growth_mb, 2),
+            window_s=round(window_s, 1),
+            timestamp=current.timestamp,
+        )
+    if rss_growth_mb >= rss_growth_warning_mb:
+        return MemoryPressureEvent(
+            level="warning",
+            reason="rss_growth",
+            rss_mb=current.rss_mb,
+            threshold_mb=rss_growth_warning_mb,
+            rss_growth_mb=round(rss_growth_mb, 2),
+            window_s=round(window_s, 1),
+            timestamp=current.timestamp,
+        )
+    return None
+
+
+def _get_state_value(state) -> str:
+    return state.value if hasattr(state, "value") else str(state)
+
+
+def _get_all_snapshots() -> dict:
+    try:
+        from yuxi.channel.runtime.manager import gateway  # 延迟导入以避免循环依赖；Python import 缓存保证无重复加载开销
+
+        return gateway.get_all_snapshots()
+    except Exception:
+        logger.warning("DiagnosticHeartbeat: get_all_snapshots failed", exc_info=True)
+        return {}
+
+
+def _count_active_sessions_from(snapshots: dict) -> int:
+    return sum(1 for s in snapshots.values() if _get_state_value(s.state) == "running")
+
+
+diagnostic_heartbeat = DiagnosticHeartbeat()
--- a/backend/package/yuxi/channel/monitoring/diagnostic_phase.py
+++ b/backend/package/yuxi/channel/monitoring/diagnostic_phase.py
@ -0,0 +1,131 @@
+import logging
+import math
+import os
+import time
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+
+logger = logging.getLogger(__name__)
+
+RECENT_PHASE_CAPACITY = 40
+
+
+@dataclass
+class PhaseSnapshot:
+    name: str
+    started_at: float
+    ended_at: float
+    duration_ms: float
+    cpu_user_ms: float
+    cpu_system_ms: float
+    cpu_total_ms: float
+    cpu_core_ratio: float
+    details: dict[str, str | int | float | bool] | None = None
+
+
+_active_phase_stack: list[dict] = []
+_recent_phases: list[PhaseSnapshot] = []
+
+
+def _round_metric(value: float, digits: int = 1) -> float:
+    if not math.isfinite(value):
+        return 0.0
+    factor = 10**digits
+    return round(value * factor) / factor
+
+
+def _push_recent(snapshot: PhaseSnapshot) -> None:
+    _recent_phases.append(snapshot)
+    if len(_recent_phases) > RECENT_PHASE_CAPACITY:
+        del _recent_phases[: len(_recent_phases) - RECENT_PHASE_CAPACITY]
+
+
+def _record_phase(snapshot: PhaseSnapshot) -> None:
+    _push_recent(snapshot)
+    try:
+        from yuxi.channel.monitoring.diagnostic_stability import record_stability_event
+
+        record_stability_event(
+            "diagnostic.phase.completed",
+            name=snapshot.name,
+            started_at=snapshot.started_at,
+            ended_at=snapshot.ended_at,
+            duration_ms=snapshot.duration_ms,
+            cpu_user_ms=snapshot.cpu_user_ms,
+            cpu_system_ms=snapshot.cpu_system_ms,
+            cpu_total_ms=snapshot.cpu_total_ms,
+            cpu_core_ratio=snapshot.cpu_core_ratio,
+        )
+    except Exception:
+        pass
+
+
+def get_current_phase() -> str | None:
+    if not _active_phase_stack:
+        return None
+    return _active_phase_stack[-1]["name"]
+
+
+def get_recent_phases(limit: int = 8) -> list[PhaseSnapshot]:
+    return list(_recent_phases[-max(0, limit) :])
+
+
+def record_diagnostic_phase(snapshot: PhaseSnapshot) -> None:
+    _record_phase(snapshot)
+
+
+def reset_diagnostic_phases_for_test() -> None:
+    _active_phase_stack.clear()
+    _recent_phases.clear()
+
+
+@asynccontextmanager
+async def diagnostic_phase(name: str, details: dict[str, str | int | float | bool] | None = None):
+    cpu_started = os.times()
+    active: dict = {
+        "name": name,
+        "started_at": time.time(),
+        "started_wall": time.perf_counter(),
+        "cpu_started_user": cpu_started.user,
+        "cpu_started_system": cpu_started.system,
+        "details": details,
+    }
+    _active_phase_stack.append(active)
+    try:
+        yield
+    finally:
+        ended_at = time.time()
+        duration_ms = _round_metric((time.perf_counter() - active["started_wall"]) * 1000)
+
+        cpu_ended = os.times()
+        cpu_user_ms = _round_metric((cpu_ended.user - active["cpu_started_user"]) * 1000)
+        cpu_system_ms = _round_metric((cpu_ended.system - active["cpu_started_system"]) * 1000)
+        cpu_total_ms = _round_metric(cpu_user_ms + cpu_system_ms)
+
+        cpu_core_ratio = _round_metric(cpu_total_ms / max(1.0, duration_ms), 3)
+
+        _active_phase_stack[:] = [e for e in _active_phase_stack if e is not active]
+
+        snapshot = PhaseSnapshot(
+            name=name,
+            started_at=active["started_at"],
+            ended_at=ended_at,
+            duration_ms=duration_ms,
+            cpu_user_ms=cpu_user_ms,
+            cpu_system_ms=cpu_system_ms,
+            cpu_total_ms=cpu_total_ms,
+            cpu_core_ratio=cpu_core_ratio,
+            details=active["details"],
+        )
+        _record_phase(snapshot)
+
+        if cpu_core_ratio > 0.9:
+            logger.debug(
+                "DiagnosticPhase '%s': %sms, cpu_user=%.1fms, cpu_system=%.1fms, cpu_total=%.1fms, ratio=%.2f",
+                name,
+                duration_ms,
+                cpu_user_ms,
+                cpu_system_ms,
+                cpu_total_ms,
+                cpu_core_ratio,
+            )
--- a/backend/package/yuxi/channel/monitoring/diagnostic_stability.py
+++ b/backend/package/yuxi/channel/monitoring/diagnostic_stability.py
@ -0,0 +1,409 @@
+import json
+import logging
+import os
+import re
+import signal
+import sys
+import time
+from collections import deque
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from yuxi.channel.security.log_sanitizer import sanitize_text
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_RING_CAPACITY = 1000
+DEFAULT_QUERY_LIMIT = 50
+MAX_BUNDLE_BYTES = 5 * 1024 * 1024
+DEFAULT_BUNDLE_RETENTION = 20
+MAX_SAFE_ERROR_MESSAGE_LENGTH = 500
+
+SAFE_REASON_CODE = re.compile(r"^[A-Za-z0-9_.:-]{1,120}$")
+REDACTED_HOSTNAME = "<redacted-hostname>"
+
+_PEM_PRIVATE_KEY_PATTERN = re.compile(
+    r"-----BEGIN\s+(?:RSA\s+)?PRIVATE\s+KEY-----[\s\S]*?"
+    r"-----END\s+(?:RSA\s+)?PRIVATE\s+KEY-----",
+)
+
+
+@dataclass
+class StabilityEvent:
+    seq: int
+    ts: float
+    type: str
+    data: dict[str, Any] = field(default_factory=dict)
+
+
+class StabilityRingBuffer:
+    def __init__(self, capacity: int = DEFAULT_RING_CAPACITY):
+        self._buffer: deque[StabilityEvent] = deque(maxlen=capacity)
+        self._seq = 0
+        self._dropped = 0
+
+    def record(self, event_type: str, **data: Any) -> None:
+        self._seq += 1
+        event = StabilityEvent(
+            seq=self._seq,
+            ts=time.time(),
+            type=event_type,
+            data=data,
+        )
+        if len(self._buffer) >= self._buffer.maxlen:
+            self._dropped += 1
+        self._buffer.append(event)
+
+    def query(
+        self,
+        limit: int = DEFAULT_QUERY_LIMIT,
+        event_type: str | None = None,
+        since_seq: int | None = None,
+    ) -> list[dict]:
+        events = list(self._buffer)
+        if event_type is not None:
+            events = [e for e in events if e.type == event_type]
+        if since_seq is not None:
+            events = [e for e in events if e.seq > since_seq]
+        events = events[-limit:]
+        return [
+            {
+                "seq": e.seq,
+                "ts": e.ts,
+                "type": e.type,
+                **e.data,
+            }
+            for e in events
+        ]
+
+    def snapshot(
+        self,
+        limit: int = DEFAULT_QUERY_LIMIT,
+        event_type: str | None = None,
+        since_seq: int | None = None,
+    ) -> dict:
+        all_events = list(self._buffer)
+        filtered = all_events
+        if event_type is not None:
+            filtered = [e for e in filtered if e.type == event_type]
+        if since_seq is not None:
+            filtered = [e for e in filtered if e.seq > since_seq]
+        events = [
+            {
+                "seq": e.seq,
+                "ts": e.ts,
+                "type": e.type,
+                **e.data,
+            }
+            for e in filtered[-limit:]
+        ]
+
+        by_type: dict[str, int] = {}
+        for e in filtered:
+            by_type[e.type] = by_type.get(e.type, 0) + 1
+
+        return {
+            "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime()),
+            "capacity": self._buffer.maxlen,
+            "count": len(filtered),
+            "dropped": self._dropped,
+            "first_seq": filtered[0].seq if filtered else None,
+            "last_seq": filtered[-1].seq if filtered else None,
+            "summary": {"by_type": by_type},
+            "events": events,
+        }
+
+    @property
+    def count(self) -> int:
+        return len(self._buffer)
+
+
+_stability_buffer = StabilityRingBuffer()
+
+_pending_stability_reason: str | None = None
+_pending_stability_error: Exception | None = None
+
+
+def _write_stability_at_exit() -> None:
+    global _pending_stability_reason, _pending_stability_error
+    if _pending_stability_reason:
+        try:
+            write_stability_bundle_for_failure(_pending_stability_reason, _pending_stability_error)
+        except Exception:
+            pass
+    _pending_stability_reason = None
+    _pending_stability_error = None
+
+
+def record_stability_event(event_type: str, **data: Any) -> None:
+    _stability_buffer.record(event_type, **data)
+
+
+def get_stability_snapshot(
+    limit: int = DEFAULT_QUERY_LIMIT,
+    event_type: str | None = None,
+    since_seq: int | None = None,
+) -> dict:
+    return _stability_buffer.snapshot(limit, event_type=event_type, since_seq=since_seq)
+
+
+def _normalize_reason(reason: str) -> str:
+    return reason if SAFE_REASON_CODE.match(reason) else "unknown"
+
+
+def _redact_sensitive_text(text: str) -> str:
+    text = sanitize_text(text)
+    text = _PEM_PRIVATE_KEY_PATTERN.sub("[REDACTED]", text)
+    return re.sub(r"\s+", " ", text).strip()
+
+
+def _extract_error_name(error: Exception) -> str | None:
+    name = getattr(error, "name", None)
+    if isinstance(name, str) and SAFE_REASON_CODE.match(name):
+        return name
+    cls_name = type(error).__name__
+    return cls_name if SAFE_REASON_CODE.match(cls_name) else None
+
+
+def _extract_error_code(error: Exception) -> str | None:
+    code = getattr(error, "code", None)
+    if isinstance(code, str) and SAFE_REASON_CODE.match(code):
+        return code
+    if isinstance(code, int) and code >= 0:
+        return str(code)
+    errno = getattr(error, "errno", None)
+    if isinstance(errno, int) and errno >= 0:
+        return str(errno)
+    return None
+
+
+def _extract_safe_error_message(error: Exception) -> str | None:
+    message = getattr(error, "message", None) or str(error)
+    if not message:
+        return None
+    sanitized = _redact_sensitive_text(message)
+    if not sanitized:
+        return None
+    if len(sanitized) > MAX_SAFE_ERROR_MESSAGE_LENGTH:
+        sanitized = sanitized[:MAX_SAFE_ERROR_MESSAGE_LENGTH] + "..."
+    return sanitized
+
+
+def _build_error_metadata(error: Exception) -> dict[str, str] | None:
+    name = _extract_error_name(error)
+    code = _extract_error_code(error)
+    message = _extract_safe_error_message(error)
+    if not name and not code and not message:
+        return None
+    result: dict[str, str] = {}
+    if name:
+        result["name"] = name
+    if code:
+        result["code"] = code
+    if message:
+        result["message"] = message
+    return result
+
+
+def _resolve_bundle_dir() -> Path:
+    return Path(os.environ.get("YUXI_STABILITY_DIR", "data/stability"))
+
+
+def _get_hostname_redacted() -> str:
+    return REDACTED_HOSTNAME
+
+
+def _get_process_uptime_ms() -> int:
+    try:
+        return int((time.time() - __import__("psutil").Process(os.getpid()).create_time()) * 1000)
+    except Exception:
+        return 0
+
+
+def _build_bundle_filename(timestamp: str, pid: int, reason: str) -> str:
+    normalized = _normalize_reason(reason)
+    return f"yuxi-stability-{timestamp}-{pid}-{normalized}.json"
+
+
+def _truncate_events_to_fit(bundle: dict, snapshot: dict, max_bytes: int) -> dict:
+    events = snapshot["events"]
+    if not events:
+        return bundle
+
+    lo, hi = 0, len(events)
+    while lo < hi:
+        mid = (lo + hi) // 2
+        test_snapshot = {**snapshot, "events": events[mid:]}
+        test_bundle = {**bundle, "snapshot": test_snapshot}
+        if len(json.dumps(test_bundle, ensure_ascii=False, indent=2).encode()) <= max_bytes:
+            hi = mid
+        else:
+            lo = mid + 1
+
+    snapshot["events"] = events[lo:]
+    bundle["snapshot"] = snapshot
+    return bundle
+
+
+def write_stability_bundle(
+    reason: str,
+    error: Exception | None = None,
+    include_empty: bool = False,
+    limit: int = DEFAULT_RING_CAPACITY,
+) -> str | None:
+    import platform
+
+    snapshot = _stability_buffer.snapshot(limit)
+    if snapshot["count"] == 0 and not error and not include_empty:
+        return None
+
+    bundle_dir = _resolve_bundle_dir()
+    bundle_dir.mkdir(parents=True, exist_ok=True)
+
+    normalized_reason = _normalize_reason(reason)
+    error_meta = _build_error_metadata(error) if error else None
+
+    bundle = {
+        "version": 1,
+        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "reason": normalized_reason,
+        "process": {
+            "pid": os.getpid(),
+            "platform": sys.platform,
+            "arch": platform.machine(),
+            "python": sys.version,
+            "uptime_ms": _get_process_uptime_ms(),
+        },
+        "host": {
+            "hostname": _get_hostname_redacted(),
+        },
+        "error": error_meta,
+        "snapshot": snapshot,
+    }
+
+    timestamp = time.strftime("%Y%m%dT%H%M%S", time.localtime())
+    filename = _build_bundle_filename(timestamp, os.getpid(), normalized_reason)
+    filepath = bundle_dir / filename
+
+    content = json.dumps(bundle, ensure_ascii=False, indent=2)
+    content_bytes = content.encode()
+    if len(content_bytes) > MAX_BUNDLE_BYTES:
+        logger.error("Stability bundle too large (%d bytes), truncating", len(content_bytes))
+        bundle = _truncate_events_to_fit(bundle, snapshot, MAX_BUNDLE_BYTES)
+        content = json.dumps(bundle, ensure_ascii=False, indent=2)
+
+    with open(filepath, "w", encoding="utf-8") as f:
+        f.write(content)
+
+    _restrict_file_permissions(filepath)
+
+    _prune_old_bundles(bundle_dir)
+    logger.warning("Stability bundle written: %s", filepath)
+    return str(filepath)
+
+
+def write_stability_bundle_for_failure(reason: str, error: Exception | None = None) -> str | None:
+    return write_stability_bundle(reason, error=error, include_empty=True)
+
+
+def _restrict_file_permissions(filepath: Path) -> None:
+    try:
+        os.chmod(filepath, 0o600)
+    except Exception:
+        pass
+
+
+def _prune_old_bundles(bundle_dir: Path) -> None:
+    try:
+        bundles = sorted(
+            bundle_dir.glob("yuxi-stability-*.json"),
+            key=lambda p: p.stat().st_mtime,
+            reverse=True,
+        )
+        for p in bundles[DEFAULT_BUNDLE_RETENTION:]:
+            try:
+                p.unlink()
+            except OSError:
+                pass
+    except Exception:
+        pass
+
+
+def list_stability_bundles(bundle_dir: Path | None = None) -> list[dict]:
+    _dir = bundle_dir or _resolve_bundle_dir()
+    try:
+        bundles = sorted(
+            _dir.glob("yuxi-stability-*.json"),
+            key=lambda p: p.stat().st_mtime,
+            reverse=True,
+        )
+        return [{"path": str(p), "mtime": p.stat().st_mtime} for p in bundles]
+    except Exception:
+        return []
+
+
+def read_stability_bundle_file(filepath: str | Path) -> dict | None:
+    filepath = Path(filepath)
+    try:
+        stat = filepath.stat()
+        if stat.st_size > MAX_BUNDLE_BYTES:
+            logger.error("Stability bundle too large: %s (%d bytes)", filepath, stat.st_size)
+            return None
+
+        with open(filepath, encoding="utf-8") as f:
+            raw = f.read()
+
+        bundle = json.loads(raw)
+        if bundle.get("version") != 1:
+            logger.error("Unsupported stability bundle version: %s", bundle.get("version"))
+            return None
+
+        reason = bundle.get("reason", "")
+        if not SAFE_REASON_CODE.match(reason):
+            bundle["reason"] = "unknown"
+
+        host = bundle.get("host", {})
+        if isinstance(host, dict):
+            bundle["host"]["hostname"] = REDACTED_HOSTNAME
+
+        return bundle
+    except (json.JSONDecodeError, OSError) as e:
+        logger.error("Failed to read stability bundle %s: %s", filepath, e)
+        return None
+
+
+def read_latest_stability_bundle(bundle_dir: Path | None = None) -> dict | None:
+    bundles = list_stability_bundles(bundle_dir)
+    if not bundles:
+        return None
+    return read_stability_bundle_file(bundles[0]["path"])
+
+
+_original_excepthook = sys.excepthook
+
+
+def install_stability_crash_hook() -> None:
+    import atexit
+
+    atexit.register(_write_stability_at_exit)
+
+    def _handler(signum, frame):
+        global _pending_stability_reason
+        _pending_stability_reason = f"signal-{signum}"
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+
+    for sig in (signal.SIGTERM, signal.SIGINT):
+        try:
+            signal.signal(sig, _handler)
+        except Exception:
+            pass
+
+    def _excepthook(exc_type, exc_value, exc_tb):
+        global _pending_stability_reason, _pending_stability_error
+        _pending_stability_reason = "unhandled-exception"
+        _pending_stability_error = exc_value
+        _original_excepthook(exc_type, exc_value, exc_tb)
+
+    sys.excepthook = _excepthook
--- a/backend/package/yuxi/channel/monitoring/event_loop_monitor.py
+++ b/backend/package/yuxi/channel/monitoring/event_loop_monitor.py
@ -0,0 +1,249 @@
+import asyncio
+import bisect
+import logging
+import os
+import time
+from collections import deque
+from dataclasses import dataclass, field
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_SAMPLE_INTERVAL_MS = 100
+DEFAULT_WINDOW_SIZE = 600
+WARN_DELAY_MS = 1000
+CRITICAL_DELAY_MS = 5000
+WARN_UTIL = 0.95
+CPU_CORE_RATIO_WARN = 0.9
+DELAY_COINCIDENCE_MS = 25
+SUSTAINED_LOAD_MIN_INTERVAL_MS = 1000
+
+
+@dataclass
+class EventLoopStats:
+    p50_ms: float = 0.0
+    p90_ms: float = 0.0
+    p99_ms: float = 0.0
+    p999_ms: float = 0.0
+    max_ms: float = 0.0
+    min_ms: float = 0.0
+    avg_ms: float = 0.0
+    utilization: float = 0.0
+    samples: int = 0
+    warnings: list[str] = field(default_factory=list)
+
+
+@dataclass
+class EventLoopHealth:
+    degraded: bool = False
+    reasons: list[str] = field(default_factory=list)
+    interval_ms: float = 0.0
+    delay_p99_ms: float = 0.0
+    delay_max_ms: float = 0.0
+    utilization: float = 0.0
+    cpu_core_ratio: float = 0.0
+
+
+class EventLoopMonitor:
+    """监控 asyncio event loop 延迟，提供 P50/P90/P99/P999 延迟 + 利用率 + CPU。
+
+    通过定期调度回调测量 event loop 处理延迟。
+    等效于 Node.js perf_hooks.monitorEventLoopDelay。
+    """
+
+    def __init__(
+        self,
+        sample_interval_ms: int = DEFAULT_SAMPLE_INTERVAL_MS,
+        window_size: int = DEFAULT_WINDOW_SIZE,
+    ):
+        self._sample_interval = sample_interval_ms / 1000.0
+        self._window_size = window_size
+        self._samples: list[float] = []
+        self._ring: deque[float] = deque()
+        self._running = False
+        self._task: asyncio.Task | None = None
+        self._last_wall_at: float = 0.0
+        self._last_process_time: float = 0.0
+        self._cpu_count = _cpu_core_count()
+        self._last_health: EventLoopHealth | None = None
+
+    async def start(self):
+        self._running = True
+        self._last_wall_at = time.perf_counter()
+        self._last_process_time = time.process_time()
+        self._task = asyncio.create_task(self._loop())
+        logger.info(
+            "EventLoopMonitor started (interval=%dms, window=%d)",
+            int(self._sample_interval * 1000),
+            self._window_size,
+        )
+
+    async def stop(self):
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None
+        logger.info("EventLoopMonitor stopped")
+
+    async def _loop(self):
+        while self._running:
+            t0 = time.perf_counter()
+            await asyncio.sleep(self._sample_interval)
+            delay_ms = max(0.0, (time.perf_counter() - t0 - self._sample_interval) * 1000)
+
+            self._ring.append(delay_ms)
+            bisect.insort(self._samples, delay_ms)
+
+            if len(self._ring) > self._window_size:
+                evicted = self._ring.popleft()
+                idx = bisect.bisect_left(self._samples, evicted)
+                self._samples.pop(idx)
+
+    def stats(self) -> EventLoopStats:
+        if not self._samples:
+            return EventLoopStats(samples=0)
+
+        sorted_samples = self._samples
+        n = len(sorted_samples)
+
+        def _pct(p: float) -> float:
+            idx = max(0, min(n - 1, int(n * p / 100.0)))
+            return sorted_samples[idx]
+
+        p50 = _pct(50)
+        p99 = _pct(99)
+        avg = sum(sorted_samples) / n
+        util = p99 / self._sample_interval / 1000.0 if self._sample_interval > 0 else 0.0
+
+        warnings: list[str] = []
+        if p99 > WARN_DELAY_MS:
+            warnings.append(f"P99延迟 {p99:.1f}ms > {WARN_DELAY_MS}ms")
+        if util > WARN_UTIL:
+            warnings.append(f"利用率 {util:.1%} > {WARN_UTIL:.0%}")
+        if p99 > CRITICAL_DELAY_MS:
+            warnings.append(f"P99延迟 {p99:.1f}ms > {CRITICAL_DELAY_MS}ms (严重)")
+
+        return EventLoopStats(
+            p50_ms=round(p50, 2),
+            p90_ms=round(_pct(90), 2),
+            p99_ms=round(p99, 2),
+            p999_ms=round(_pct(99.9), 2),
+            max_ms=round(sorted_samples[-1], 2),
+            min_ms=round(sorted_samples[0], 2),
+            avg_ms=round(avg, 2),
+            utilization=round(util, 4),
+            samples=n,
+            warnings=warnings,
+        )
+
+    @property
+    def last_health(self) -> EventLoopHealth | None:
+        return self._last_health
+
+    def health(self) -> EventLoopHealth:
+        """评估 event loop 当前健康状态。
+
+        注意：此方法会更新内部时间戳和 CPU 计数器，作为后续调用的基准。
+        如需只读访问上次评估结果，请使用 last_health 属性。
+        """
+        now = time.perf_counter()
+        interval_ms = max(1.0, (now - self._last_wall_at) * 1000)
+        stats = self.stats()
+
+        if stats.samples == 0:
+            return EventLoopHealth()
+
+        delay_p99_ms = stats.p99_ms
+        delay_max_ms = stats.max_ms
+        has_delay_warning = delay_p99_ms >= WARN_DELAY_MS or delay_max_ms >= WARN_DELAY_MS
+
+        if not has_delay_warning and interval_ms < SUSTAINED_LOAD_MIN_INTERVAL_MS:
+            if self._last_health is not None:
+                return self._last_health
+            return EventLoopHealth(
+                interval_ms=round(interval_ms, 1),
+                delay_p99_ms=delay_p99_ms,
+                delay_max_ms=delay_max_ms,
+                utilization=stats.utilization,
+            )
+
+        current_process_time = time.process_time()
+        cpu_delta = current_process_time - self._last_process_time
+        cpu_core_ratio = round(cpu_delta / max(interval_ms / 1000, 0.001), 3)
+
+        reasons = _classify_health_reasons(
+            interval_ms=interval_ms,
+            delay_p99_ms=delay_p99_ms,
+            delay_max_ms=delay_max_ms,
+            utilization=stats.utilization,
+            cpu_core_ratio=cpu_core_ratio,
+        )
+
+        health = EventLoopHealth(
+            degraded=len(reasons) > 0,
+            reasons=reasons,
+            interval_ms=round(interval_ms, 1),
+            delay_p99_ms=delay_p99_ms,
+            delay_max_ms=delay_max_ms,
+            utilization=stats.utilization,
+            cpu_core_ratio=cpu_core_ratio,
+        )
+
+        self._last_wall_at = now
+        self._last_process_time = current_process_time
+        self._last_health = health
+
+        _push_event_loop_metrics(health)
+
+        return health
+
+
+def _classify_health_reasons(
+    *,
+    interval_ms: float,
+    delay_p99_ms: float,
+    delay_max_ms: float,
+    utilization: float,
+    cpu_core_ratio: float,
+) -> list[str]:
+    reasons: list[str] = []
+
+    if delay_p99_ms >= WARN_DELAY_MS or delay_max_ms >= WARN_DELAY_MS:
+        reasons.append("event_loop_delay")
+
+    if interval_ms < SUSTAINED_LOAD_MIN_INTERVAL_MS:
+        return reasons
+
+    has_delay_co_evidence = delay_p99_ms >= DELAY_COINCIDENCE_MS or delay_max_ms >= DELAY_COINCIDENCE_MS
+    if not has_delay_co_evidence:
+        return reasons
+
+    if utilization >= WARN_UTIL:
+        reasons.append("event_loop_utilization")
+    if cpu_core_ratio >= CPU_CORE_RATIO_WARN:
+        reasons.append("cpu")
+
+    return reasons
+
+
+def _cpu_core_count() -> int:
+    try:
+        return os.cpu_count() or 1
+    except Exception:
+        return 1
+
+
+def _push_event_loop_metrics(health: EventLoopHealth):
+    try:
+        from yuxi.channel.monitoring.metrics import event_loop_delay_p99, event_loop_utilization
+
+        event_loop_delay_p99.set(health.delay_p99_ms)
+        event_loop_utilization.set(health.utilization)
+    except Exception:
+        pass
+
+
+event_loop_monitor = EventLoopMonitor()
--- a/backend/package/yuxi/channel/monitoring/event_loop_ready.py
+++ b/backend/package/yuxi/channel/monitoring/event_loop_ready.py
@ -0,0 +1,139 @@
+import asyncio
+import logging
+import time
+from dataclasses import dataclass
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_MAX_WAIT_MS = 10_000
+DEFAULT_INTERVAL_MS = 1
+DEFAULT_DRIFT_THRESHOLD_MS = 200
+DEFAULT_CONSECUTIVE_READY_CHECKS = 2
+
+MAX_SAFE_TIMEOUT_DELAY_MS = 2_147_483_647
+
+
+@dataclass
+class EventLoopReadyResult:
+    ready: bool
+    elapsed_ms: float
+    max_drift_ms: float
+    checks: int
+    aborted: bool
+
+
+def _resolve_positive_int(value: float | int | None, fallback: int) -> int:
+    if value is None:
+        return fallback
+    candidate = int(value)
+    return max(1, candidate) if candidate > 0 else fallback
+
+
+def _resolve_safe_delay_ms(delay_ms: float, min_ms: int = 1) -> int:
+    candidate = _resolve_positive_int(delay_ms, min_ms)
+    return min(MAX_SAFE_TIMEOUT_DELAY_MS, candidate)
+
+
+async def wait_for_event_loop_ready(
+    *,
+    max_wait_ms: float = DEFAULT_MAX_WAIT_MS,
+    interval_ms: float = DEFAULT_INTERVAL_MS,
+    drift_threshold_ms: float = DEFAULT_DRIFT_THRESHOLD_MS,
+    consecutive_ready_checks: int = DEFAULT_CONSECUTIVE_READY_CHECKS,
+    cancel_event: asyncio.Event | None = None,
+) -> EventLoopReadyResult:
+    """等待事件循环就绪，通过 setTimeout 漂移检测判断。
+
+    在 asyncio 中通过 call_later 调度回调测量实际延迟与预期延迟之间的漂移。
+    当连续 consecutive_ready_checks 次漂移在阈值内时认为事件循环已就绪。
+
+    可通过 cancel_event 参数取消等待，取消后返回 aborted=True 的结果。
+    """
+    max_wait_ms_val = float(_resolve_safe_delay_ms(max_wait_ms))
+    interval_ms_val = float(_resolve_positive_int(interval_ms, DEFAULT_INTERVAL_MS))
+    drift_threshold_ms_val = float(_resolve_positive_int(drift_threshold_ms, DEFAULT_DRIFT_THRESHOLD_MS))
+    consecutive_ready_checks_val = _resolve_positive_int(consecutive_ready_checks, DEFAULT_CONSECUTIVE_READY_CHECKS)
+
+    started_at = time.perf_counter()
+    ready_checks = 0
+    checks = 0
+    max_drift_ms = 0.0
+
+    loop = asyncio.get_running_loop()
+    fut: asyncio.Future[EventLoopReadyResult] = loop.create_future()
+    timer_handle: asyncio.TimerHandle | None = None
+    settled = False
+
+    def clear_timer():
+        nonlocal timer_handle
+        if timer_handle is not None:
+            timer_handle.cancel()
+            timer_handle = None
+
+    def finish(ready: bool, aborted: bool = False):
+        nonlocal settled, timer_handle
+        if settled:
+            return
+        settled = True
+        clear_timer()
+        elapsed_ms = max(0.0, (time.perf_counter() - started_at) * 1000)
+        fut.set_result(
+            EventLoopReadyResult(
+                ready=ready,
+                elapsed_ms=round(elapsed_ms, 1),
+                max_drift_ms=round(max_drift_ms, 2),
+                checks=checks,
+                aborted=aborted,
+            )
+        )
+
+    def schedule_next():
+        nonlocal ready_checks, checks, max_drift_ms, timer_handle
+
+        remaining_ms = max_wait_ms_val - max(0.0, (time.perf_counter() - started_at) * 1000)
+        if remaining_ms <= 0:
+            finish(False)
+            return
+
+        delay_s = min(interval_ms_val, remaining_ms) / 1000
+        scheduled_at = time.perf_counter()
+
+        def on_timer():
+            nonlocal ready_checks, checks, max_drift_ms, timer_handle
+            checks += 1
+            drift_ms = max(0.0, (time.perf_counter() - scheduled_at) * 1000 - (delay_s * 1000))
+            max_drift_ms = max(max_drift_ms, drift_ms)
+
+            if drift_ms > drift_threshold_ms_val:
+                ready_checks = 0
+            else:
+                ready_checks += 1
+
+            if ready_checks >= consecutive_ready_checks_val:
+                finish(True)
+                return
+
+            schedule_next()
+
+        timer_handle = loop.call_later(delay_s, on_timer)
+
+    schedule_next()
+
+    cancel_watch_task: asyncio.Task | None = None
+    if cancel_event is not None:
+        if cancel_event.is_set():
+            finish(False, True)
+        else:
+
+            async def _watch_cancel():
+                await cancel_event.wait()
+                finish(False, True)
+
+            cancel_watch_task = asyncio.create_task(_watch_cancel())
+
+    try:
+        return await fut
+    finally:
+        clear_timer()
+        if cancel_watch_task is not None:
+            cancel_watch_task.cancel()
--- a/backend/package/yuxi/channel/monitoring/health_monitor.py
+++ b/backend/package/yuxi/channel/monitoring/health_monitor.py
@ -0,0 +1,468 @@
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import TypedDict
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_CHECK_INTERVAL = 300
+DEFAULT_STARTUP_GRACE = 60
+DEFAULT_CONNECTION_GRACE = 120
+DEFAULT_COOLDOWN_CYCLES = 2
+DEFAULT_MAX_RESTARTS_PER_HOUR = 10
+DEFAULT_STALE_EVENT_THRESHOLD = 1800
+BUSY_ACTIVITY_STALE_THRESHOLD = 1500
+RECONNECT_GAVE_UP_THRESHOLD = 10
+
+
+class ChannelSnapshotData(TypedDict, total=False):
+    enabled: bool | None
+    configured: bool | None
+    running: bool
+    connected: bool
+    busy: bool
+    active_runs: int
+    last_run_activity_at: float | None
+    last_transport_activity_at: float | None
+    last_start_at: float | None
+    reconnect_attempts: int
+    restart_pending: bool | None
+    state: str
+
+
+@dataclass
+class ChannelHealthEvaluation:
+    healthy: bool
+    reason: str
+    detail: str | None = None
+
+
+@dataclass
+class ChannelHealthEvent:
+    channel_type: str
+    account_id: str
+    from_state: str
+    to_state: str
+    timestamp: float = field(default_factory=time.time)
+    detail: str | None = None
+
+    def to_dict(self) -> dict:
+        return {
+            "channel_type": self.channel_type,
+            "account_id": self.account_id,
+            "from_state": self.from_state,
+            "to_state": self.to_state,
+            "timestamp": self.timestamp,
+            "detail": self.detail,
+        }
+
+
+@dataclass
+class ChannelHealthMetrics:
+    checks_total: int = 0
+    transitions: list[ChannelHealthEvent] = field(default_factory=list)
+    current_status: str = "unknown"
+    last_check_at: float = 0.0
+    unhealthy_channels: list[str] = field(default_factory=list)
+    degraded_channels: list[str] = field(default_factory=list)
+    evaluation_details: dict[str, dict] = field(default_factory=dict)
+
+    def to_dict(self) -> dict:
+        return {
+            "checks_total": self.checks_total,
+            "current_status": self.current_status,
+            "last_check_at": self.last_check_at,
+            "unhealthy_channels": self.unhealthy_channels,
+            "degraded_channels": self.degraded_channels,
+            "evaluation_details": self.evaluation_details,
+            "recent_transitions": [t.to_dict() for t in self.transitions[-20:]],
+        }
+
+
+def _safe_finite_timestamp(value: float | None) -> float | None:
+    if value is None:
+        return None
+    if not isinstance(value, (int, float)) or value != value:
+        return None
+    if value == float("inf") or value == float("-inf"):
+        return None
+    return value
+
+
+def _safe_active_runs(value: object) -> int:
+    if isinstance(value, (int, float)):
+        if value != value or value == float("inf") or value == float("-inf"):
+            return 0
+        return max(0, int(value))
+    return 0
+
+
+def evaluate_channel_health(
+    *,
+    enabled: bool | None = None,
+    configured: bool | None = None,
+    running: bool = False,
+    connected: bool = False,
+    busy: bool = False,
+    active_runs: int = 0,
+    last_run_activity_at: float | None = None,
+    last_transport_activity_at: float | None = None,
+    last_start_at: float | None = None,
+    reconnect_attempts: int = 0,
+    now: float | None = None,
+    stale_event_threshold: float = DEFAULT_STALE_EVENT_THRESHOLD,
+    connect_grace: float = DEFAULT_CONNECTION_GRACE,
+) -> ChannelHealthEvaluation:
+    if now is None:
+        now = time.monotonic()
+
+    if enabled is False or configured is False:
+        return ChannelHealthEvaluation(healthy=True, reason="unmanaged")
+
+    if not running:
+        reason = "gave-up" if reconnect_attempts >= RECONNECT_GAVE_UP_THRESHOLD else "not-running"
+        return ChannelHealthEvaluation(
+            healthy=False,
+            reason=reason,
+            detail=f"reconnect_attempts={reconnect_attempts}",
+        )
+
+    safe_active_runs = _safe_active_runs(active_runs)
+    is_busy = busy or safe_active_runs > 0
+
+    safe_last_start_at = _safe_finite_timestamp(last_start_at)
+    safe_last_run_activity_at = _safe_finite_timestamp(last_run_activity_at)
+    safe_last_transport_activity_at = _safe_finite_timestamp(last_transport_activity_at)
+
+    if is_busy:
+        busy_state_initialized_for_lifecycle = safe_last_start_at is None or (
+            safe_last_run_activity_at is not None and safe_last_run_activity_at >= safe_last_start_at
+        )
+        if busy_state_initialized_for_lifecycle:
+            run_activity_age = (
+                float("inf") if safe_last_run_activity_at is None else max(0.0, now - safe_last_run_activity_at)
+            )
+            if run_activity_age < BUSY_ACTIVITY_STALE_THRESHOLD:
+                return ChannelHealthEvaluation(healthy=True, reason="busy")
+            return ChannelHealthEvaluation(
+                healthy=False,
+                reason="stuck",
+                detail=f"run_activity_age={run_activity_age:.0f}s",
+            )
+
+    if safe_last_start_at is not None:
+        up_duration = now - safe_last_start_at
+        if up_duration < connect_grace:
+            return ChannelHealthEvaluation(healthy=True, reason="startup-connect-grace")
+
+    if connected is False:
+        return ChannelHealthEvaluation(healthy=False, reason="disconnected")
+
+    if connected and safe_last_transport_activity_at is not None:
+        if safe_last_start_at is not None and safe_last_transport_activity_at < safe_last_start_at:
+            lifecycle_gap = max(0.0, now - safe_last_start_at)
+            if lifecycle_gap <= stale_event_threshold:
+                return ChannelHealthEvaluation(healthy=True, reason="healthy")
+            return ChannelHealthEvaluation(healthy=False, reason="stale-socket")
+
+        event_age = now - safe_last_transport_activity_at
+        if event_age > stale_event_threshold:
+            return ChannelHealthEvaluation(
+                healthy=False,
+                reason="stale-socket",
+                detail=f"event_age={event_age:.0f}s",
+            )
+
+    return ChannelHealthEvaluation(healthy=True, reason="healthy")
+
+
+def resolve_restart_reason(
+    snapshot: ChannelSnapshotData,
+    evaluation: ChannelHealthEvaluation,
+) -> str:
+    if evaluation.reason == "stale-socket":
+        return "stale-socket"
+    if evaluation.reason == "not-running":
+        reconnect_attempts = snapshot.get("reconnect_attempts", 0)
+        return "gave-up" if reconnect_attempts >= RECONNECT_GAVE_UP_THRESHOLD else "stopped"
+    if evaluation.reason == "disconnected":
+        return "disconnected"
+    if evaluation.reason == "stuck":
+        return "stuck"
+    return evaluation.reason
+
+
+@dataclass
+class _RestartRecord:
+    last_restart_at: float = 0.0
+    restarts_this_hour: list[float] = field(default_factory=list)
+
+    def prune(self, now: float) -> None:
+        cutoff = now - 3600
+        self.restarts_this_hour = [t for t in self.restarts_this_hour if t > cutoff]
+
+
+class ChannelHealthMonitor:
+    """频道健康自动监控器：独立后台轮询，5min 间隔自动检查。
+
+    特性：
+    - 启动宽限 60s（启动期间不告警）
+    - 连接宽限 120s（新建连接等待稳定）
+    - 冷却 2 周期（时间窗口冷却，避免抖动恢复触发频繁重启）
+    - 每 channel 每小时重启上限 10 次（独立计数，不互相影响）
+    - Stale-socket / stuck / gave-up 检测
+    - 检测到不健康的 channel 后自动重启（按 stop → start 流程）
+    - 自动跳过手动停止的 channel（不做重启）
+    """
+
+    def __init__(
+        self,
+        check_interval: int = DEFAULT_CHECK_INTERVAL,
+        startup_grace: int = DEFAULT_STARTUP_GRACE,
+        connection_grace: int = DEFAULT_CONNECTION_GRACE,
+        cooldown_cycles: int = DEFAULT_COOLDOWN_CYCLES,
+        max_restarts_per_hour: int = DEFAULT_MAX_RESTARTS_PER_HOUR,
+        stale_event_threshold: int = DEFAULT_STALE_EVENT_THRESHOLD,
+    ):
+        self._check_interval = check_interval
+        self._startup_grace = startup_grace
+        self._connection_grace = connection_grace
+        self._cooldown_cycles = cooldown_cycles
+        self._max_restarts_per_hour = max_restarts_per_hour
+        self._stale_event_threshold = stale_event_threshold
+        self._running = False
+        self._check_in_flight = False
+        self._task: asyncio.Task | None = None
+        self._started_at: float = 0.0
+        self._previous_states: dict[str, str] = {}
+        self._restart_records: dict[str, _RestartRecord] = {}
+        self._metrics = ChannelHealthMetrics()
+
+    @property
+    def metrics(self) -> ChannelHealthMetrics:
+        return self._metrics
+
+    async def start(self):
+        self._running = True
+        self._started_at = time.monotonic()
+        self._task = asyncio.create_task(self._loop())
+        logger.info(
+            "ChannelHealthMonitor started (interval=%ds, startup_grace=%ds, "
+            "connection_grace=%ds, cooldown=%d cycles, max_restarts=%d/h/channel, "
+            "stale_event_threshold=%ds)",
+            self._check_interval,
+            self._startup_grace,
+            self._connection_grace,
+            self._cooldown_cycles,
+            self._max_restarts_per_hour,
+            self._stale_event_threshold,
+        )
+
+    async def stop(self):
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None
+        logger.info("ChannelHealthMonitor stopped")
+
+    async def _loop(self):
+        while self._running:
+            await asyncio.sleep(self._check_interval)
+            try:
+                await self._check_and_alert()
+            except Exception:
+                logger.exception("ChannelHealthMonitor: check failed")
+
+    async def _check_and_alert(self):
+        if self._check_in_flight:
+            return
+        self._check_in_flight = True
+
+        try:
+            await self._do_check()
+        finally:
+            self._check_in_flight = False
+
+    async def _do_check(self):
+        now = time.monotonic()
+        elapsed = now - self._started_at
+
+        if elapsed < self._startup_grace:
+            logger.debug(
+                "ChannelHealthMonitor: startup grace period (%.0fs / %ds)",
+                elapsed,
+                self._startup_grace,
+            )
+            return
+
+        from yuxi.channel.runtime.manager import gateway  # 延迟导入以避免循环依赖；Python import 缓存保证无重复加载开销
+
+        report = gateway.get_health()
+        self._metrics.checks_total += 1
+        self._metrics.last_check_at = now
+        self._metrics.current_status = report.status
+
+        cooldown_window = self._cooldown_cycles * self._check_interval
+
+        unhealthy: list[str] = []
+        degraded: list[str] = []
+        evaluation_details: dict[str, dict] = {}
+
+        for key, snap_data in report.channels.items():
+            parts = key.split(":", 1)
+            if len(parts) != 2:
+                logger.warning("ChannelHealthMonitor: invalid channel key format: %s", key)
+                continue
+            channel_type, account_id = parts
+            state = snap_data.get("state", "stopped")
+            prev_state = self._previous_states.get(key, "stopped")
+
+            evaluation = evaluate_channel_health(
+                enabled=snap_data.get("enabled"),
+                configured=snap_data.get("configured"),
+                running=snap_data.get("running", False),
+                connected=snap_data.get("connected", False),
+                busy=snap_data.get("busy", False),
+                active_runs=snap_data.get("active_runs", 0),
+                last_run_activity_at=snap_data.get("last_run_activity_at"),
+                last_transport_activity_at=snap_data.get("last_transport_activity_at"),
+                last_start_at=snap_data.get("last_start_at"),
+                reconnect_attempts=snap_data.get("reconnect_attempts", 0),
+                now=now,
+                stale_event_threshold=self._stale_event_threshold,
+                connect_grace=self._connection_grace,
+            )
+
+            evaluation_details[key] = {
+                "healthy": evaluation.healthy,
+                "reason": evaluation.reason,
+                "detail": evaluation.detail,
+                "restart_reason": resolve_restart_reason(snap_data, evaluation) if not evaluation.healthy else None,
+            }
+
+            if prev_state != state:
+                event = ChannelHealthEvent(
+                    channel_type=channel_type,
+                    account_id=account_id,
+                    from_state=prev_state,
+                    to_state=state,
+                    detail=snap_data.get("last_error"),
+                )
+                self._metrics.transitions.append(event)
+                self._previous_states[key] = state
+
+                logger.info(
+                    "ChannelHealthMonitor: %s:%s %s → %s (health=%s)",
+                    channel_type,
+                    account_id,
+                    prev_state,
+                    state,
+                    evaluation.reason,
+                )
+
+            if not evaluation.healthy and evaluation.reason not in ("unmanaged", "startup-connect-grace"):
+                unhealthy.append(key)
+            elif evaluation.reason == "startup-connect-grace":
+                degraded.append(key)
+
+            if evaluation.healthy or evaluation.reason in (
+                "startup-connect-grace",
+                "unmanaged",
+            ):
+                continue
+
+            if gateway.is_manually_stopped(channel_type, account_id):
+                logger.debug("ChannelHealthMonitor: %s 手动停止中，跳过自动恢复", key)
+                continue
+
+            restart_reason = resolve_restart_reason(snap_data, evaluation)
+            logger.warning(
+                "ChannelHealthMonitor: %s 不健康 (reason=%s, restart_reason=%s)，准备自动恢复",
+                key,
+                evaluation.reason,
+                restart_reason,
+            )
+
+            record = self._restart_records.get(key)
+            if record is None:
+                record = _RestartRecord()
+                self._restart_records[key] = record
+
+            if now - record.last_restart_at <= cooldown_window:
+                continue
+
+            record.prune(now)
+            if len(record.restarts_this_hour) >= self._max_restarts_per_hour:
+                logger.warning(
+                    "ChannelHealthMonitor: %s 已达到每小时重启上限 (%d/%d)，跳过本次自动恢复",
+                    key,
+                    len(record.restarts_this_hour),
+                    self._max_restarts_per_hour,
+                )
+                continue
+
+            record.last_restart_at = now
+            record.restarts_this_hour.append(now)
+
+            logger.info(
+                "ChannelHealthMonitor: %s 触发自动恢复 (reason=%s)",
+                key,
+                restart_reason,
+            )
+            try:
+                if snap_data.get("running"):
+                    await gateway.stop_channel(channel_type, account_id)
+                await gateway.start_channel(channel_type, account_id)
+            except Exception:
+                logger.exception("ChannelHealthMonitor: %s 自动恢复执行失败", key)
+
+        self._metrics.unhealthy_channels = unhealthy
+        self._metrics.degraded_channels = degraded
+        self._metrics.evaluation_details = evaluation_details
+
+        known_keys = set(report.channels.keys())
+        for key in list(self._restart_records.keys()):
+            if key not in known_keys:
+                del self._restart_records[key]
+
+        if unhealthy:
+            logger.error(
+                "ChannelHealthMonitor: %d unhealthy channels: %s",
+                len(unhealthy),
+                ", ".join(unhealthy),
+            )
+            _maybe_send_alert(unhealthy, evaluation_details)
+        if degraded:
+            logger.warning(
+                "ChannelHealthMonitor: %d degraded channels: %s",
+                len(degraded),
+                ", ".join(degraded),
+            )
+
+    def record_restart(self, key: str):
+        record = self._restart_records.get(key)
+        if record is None:
+            record = _RestartRecord()
+            self._restart_records[key] = record
+        record.last_restart_at = time.monotonic()
+        record.restarts_this_hour.append(time.monotonic())
+
+
+async def _maybe_send_alert(unhealthy: list[str], details: dict):
+    try:
+        from yuxi.channel.monitoring.metrics import send_alert_webhook
+
+        await send_alert_webhook(
+            f"ChannelHealthMonitor: {len(unhealthy)} unhealthy channels",
+            {"channels": unhealthy, "details": details},
+        )
+    except Exception:
+        pass
+
+
+channel_health_monitor = ChannelHealthMonitor()
--- a/backend/package/yuxi/channel/monitoring/metrics.py
+++ b/backend/package/yuxi/channel/monitoring/metrics.py
@ -0,0 +1,134 @@
+import logging
+import os
+import time
+from collections import deque
+from dataclasses import dataclass, field
+
+logger = logging.getLogger(__name__)
+
+HISTORY_CAPACITY = 60
+
+
+@dataclass
+class _MetricPoint:
+    ts: float
+    value: float
+    labels: dict[str, str] = field(default_factory=dict)
+
+
+class _MetricFamily:
+    def __init__(self, name: str, help_text: str, metric_type: str):
+        self.name = name
+        self.help = help_text
+        self.type = metric_type
+        self._buffer: deque[_MetricPoint] = deque(maxlen=HISTORY_CAPACITY)
+        self._last_value: float | None = None
+        self._last_labels: dict[str, str] | None = None
+
+    def set(self, value: float, labels: dict[str, str] | None = None):
+        self._last_value = value
+        self._last_labels = labels
+        self._buffer.append(_MetricPoint(ts=time.time(), value=value, labels=labels or {}))
+
+    @property
+    def value(self) -> float | None:
+        return self._last_value
+
+    def history(self) -> list[dict]:
+        return [{"ts": p.ts, "value": p.value, "labels": p.labels} for p in self._buffer]
+
+    def prometheus_line(self) -> str:
+        if self._last_value is None:
+            return ""
+        labels = self._last_labels or {}
+        label_str = "{" + ",".join(f'{k}="{v}"' for k, v in labels.items()) + "}" if labels else ""
+        return f"{self.name}{label_str} {self._last_value}\n"
+
+
+class MetricsRegistry:
+    def __init__(self):
+        self._metrics: dict[str, _MetricFamily] = {}
+
+    def gauge(self, name: str, help_text: str) -> _MetricFamily:
+        if name not in self._metrics:
+            self._metrics[name] = _MetricFamily(name, help_text, "gauge")
+        return self._metrics[name]
+
+    def collect_prometheus(self) -> str:
+        lines = []
+        seen_help: set[str] = set()
+        for m in self._metrics.values():
+            if m.name not in seen_help:
+                lines.append(f"# HELP {m.name} {m.help}")
+                lines.append(f"# TYPE {m.name} {m.type}")
+                seen_help.add(m.name)
+            line = m.prometheus_line()
+            if line:
+                lines.append(line)
+        return "\n".join(lines) + "\n"
+
+    def history(self, name: str) -> list[dict]:
+        m = self._metrics.get(name)
+        return m.history() if m else []
+
+    def all_history(self) -> dict[str, list[dict]]:
+        return {name: m.history() for name, m in self._metrics.items()}
+
+
+metrics_registry = MetricsRegistry()
+
+event_loop_delay_p99 = metrics_registry.gauge(
+    "yuxi_event_loop_delay_p99_ms",
+    "Event loop P99 delay in milliseconds",
+)
+event_loop_utilization = metrics_registry.gauge(
+    "yuxi_event_loop_utilization",
+    "Event loop utilization ratio",
+)
+memory_rss_mb = metrics_registry.gauge(
+    "yuxi_memory_rss_mb",
+    "Process RSS memory in MB",
+)
+memory_gc_objects = metrics_registry.gauge(
+    "yuxi_memory_gc_objects",
+    "Number of objects tracked by GC",
+)
+active_sessions = metrics_registry.gauge(
+    "yuxi_active_sessions",
+    "Number of active channel sessions",
+)
+stuck_sessions = metrics_registry.gauge(
+    "yuxi_stuck_sessions",
+    "Number of stuck channel sessions",
+)
+channel_health = metrics_registry.gauge(
+    "yuxi_channel_health",
+    "Channel health status (1=healthy, 0=unhealthy)",
+)
+active_tasks = metrics_registry.gauge(
+    "yuxi_active_tasks",
+    "Number of active asyncio tasks",
+)
+
+
+async def send_alert_webhook(title: str, details: dict | None = None):
+    webhook_url = os.environ.get("YUXI_ALERT_WEBHOOK_URL")
+    if not webhook_url:
+        return
+
+    payload = {
+        "title": title,
+        "timestamp": time.time(),
+    }
+    if details:
+        payload["details"] = details
+
+    try:
+        import aiohttp
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(webhook_url, json=payload, timeout=aiohttp.ClientTimeout(total=5)) as resp:
+                if resp.status >= 400:
+                    logger.warning("Alert webhook failed: HTTP %s", resp.status)
+    except Exception:
+        logger.warning("Alert webhook delivery failed", exc_info=True)
--- a/backend/package/yuxi/channel/monitoring/readiness.py
+++ b/backend/package/yuxi/channel/monitoring/readiness.py
@ -0,0 +1,204 @@
+import logging
+import time
+from collections.abc import Callable
+from dataclasses import dataclass, field
+
+from yuxi.channel.monitoring.event_loop_monitor import EventLoopHealth
+from yuxi.channel.monitoring.health_monitor import (
+    DEFAULT_CONNECTION_GRACE,
+    DEFAULT_STALE_EVENT_THRESHOLD,
+    evaluate_channel_health,
+)
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_READINESS_CACHE_MS = 1000.0
+
+
+@dataclass
+class ChannelReadiness:
+    key: str
+    healthy: bool
+    reason: str
+    detail: str | None = None
+
+
+@dataclass
+class ReadinessResult:
+    ready: bool
+    channels: list[ChannelReadiness]
+    failing: list[str] = field(default_factory=list)
+    uptime_ms: float = 0.0
+    event_loop_degraded: bool = False
+    event_loop_reasons: list[str] = field(default_factory=list)
+    event_loop: EventLoopHealth | None = None
+    checked_at: float = 0.0
+    cache_ttl_ms: float = 1000.0
+
+
+def _should_ignore_readiness_failure(
+    evaluation_reason: str,
+    snap_data: dict,
+) -> bool:
+    if evaluation_reason == "unmanaged":
+        return True
+    if evaluation_reason == "stale-socket":
+        return True
+    if evaluation_reason == "not-running" and snap_data.get("restart_pending") is True:
+        return True
+    return False
+
+
+class ReadinessChecker:
+    def __init__(
+        self,
+        stale_event_threshold: float = DEFAULT_STALE_EVENT_THRESHOLD,
+        connect_grace: float = DEFAULT_CONNECTION_GRACE,
+        cache_ttl_ms: float = DEFAULT_READINESS_CACHE_MS,
+        started_at: float | None = None,
+        get_startup_pending: Callable[[], bool] | None = None,
+        get_startup_pending_reason: Callable[[], str] | None = None,
+        get_event_loop_health: Callable[[], EventLoopHealth | None] | None = None,
+        should_skip_channel_readiness: Callable[[], bool] | None = None,
+    ):
+        self._stale_event_threshold = stale_event_threshold
+        self._connect_grace = connect_grace
+        self._cache_ttl_ms = max(0.0, cache_ttl_ms)
+        self._started_at = started_at if started_at is not None else time.monotonic()
+        self._get_startup_pending = get_startup_pending
+        self._get_startup_pending_reason = get_startup_pending_reason
+        self._get_event_loop_health = get_event_loop_health
+        self._should_skip_channel_readiness = should_skip_channel_readiness
+        self._last_result: ReadinessResult | None = None
+        self._last_check_at: float = 0.0
+
+    async def check(self, *, force: bool = False) -> ReadinessResult:
+        now = time.monotonic()
+        uptime_ms = (now - self._started_at) * 1000
+
+        if self._get_startup_pending and self._get_startup_pending():
+            reason = self._get_startup_pending_reason() if self._get_startup_pending_reason else "startup-sidecars"
+            result = ReadinessResult(
+                ready=False,
+                channels=[],
+                failing=[reason],
+                uptime_ms=uptime_ms,
+                checked_at=now,
+                cache_ttl_ms=self._cache_ttl_ms,
+            )
+            result = self._attach_event_loop(result)
+            return result
+
+        if self._should_skip_channel_readiness and self._should_skip_channel_readiness():
+            result = ReadinessResult(
+                ready=True,
+                channels=[],
+                failing=[],
+                uptime_ms=uptime_ms,
+                checked_at=now,
+                cache_ttl_ms=self._cache_ttl_ms,
+            )
+            result = self._attach_event_loop(result)
+            return result
+
+        if not force and self._last_result is not None:
+            elapsed = (now - self._last_check_at) * 1000
+            if elapsed < self._cache_ttl_ms:
+                cached = ReadinessResult(
+                    ready=self._last_result.ready,
+                    channels=list(self._last_result.channels),
+                    failing=list(self._last_result.failing),
+                    uptime_ms=uptime_ms,
+                    event_loop_degraded=self._last_result.event_loop_degraded,
+                    event_loop_reasons=list(self._last_result.event_loop_reasons),
+                    event_loop=self._last_result.event_loop,
+                    checked_at=self._last_result.checked_at,
+                    cache_ttl_ms=self._last_result.cache_ttl_ms,
+                )
+                cached = self._attach_event_loop(cached)
+                return cached
+
+        channel_evaluations: list[ChannelReadiness] = []
+        failing: list[str] = []
+
+        try:
+            from yuxi.channel.runtime.manager import gateway
+
+            report = gateway.get_health()
+            for key, snap_data in report.channels.items():
+                evaluation = evaluate_channel_health(
+                    enabled=snap_data.get("enabled"),
+                    configured=snap_data.get("configured"),
+                    running=snap_data.get("running", False),
+                    connected=snap_data.get("connected", False),
+                    busy=snap_data.get("busy", False),
+                    active_runs=snap_data.get("active_runs", 0),
+                    last_run_activity_at=snap_data.get("last_run_activity_at"),
+                    last_transport_activity_at=snap_data.get("last_transport_activity_at"),
+                    last_start_at=snap_data.get("last_start_at"),
+                    reconnect_attempts=snap_data.get("reconnect_attempts", 0),
+                    now=now,
+                    stale_event_threshold=self._stale_event_threshold,
+                    connect_grace=self._connect_grace,
+                )
+                if evaluation.reason == "unmanaged":
+                    continue
+
+                effective_healthy = (
+                    True
+                    if not evaluation.healthy and _should_ignore_readiness_failure(evaluation.reason, snap_data)
+                    else evaluation.healthy
+                )
+
+                channel_evaluations.append(
+                    ChannelReadiness(
+                        key=key,
+                        healthy=effective_healthy,
+                        reason=evaluation.reason,
+                        detail=evaluation.detail,
+                    )
+                )
+
+                if not effective_healthy:
+                    failing.append(key)
+        except Exception as e:
+            logger.warning("ReadinessChecker: failed to evaluate channels: %s", e)
+            failing.append("evaluation-error")
+
+        all_ready = len(failing) == 0
+
+        result = ReadinessResult(
+            ready=all_ready,
+            channels=channel_evaluations,
+            failing=failing,
+            uptime_ms=uptime_ms,
+            checked_at=now,
+            cache_ttl_ms=self._cache_ttl_ms,
+        )
+        result = self._attach_event_loop(result)
+        if result.event_loop_degraded:
+            all_ready = False
+            result.ready = all_ready
+
+        self._last_result = result
+        self._last_check_at = now
+        return result
+
+    def _attach_event_loop(self, result: ReadinessResult) -> ReadinessResult:
+        try:
+            if self._get_event_loop_health:
+                event_loop = self._get_event_loop_health()
+            else:
+                from yuxi.channel.monitoring.event_loop_monitor import event_loop_monitor
+
+                event_loop = event_loop_monitor.health()
+            if event_loop is not None:
+                result.event_loop = event_loop
+                result.event_loop_degraded = event_loop.degraded
+                result.event_loop_reasons = list(event_loop.reasons)
+        except Exception as e:
+            logger.debug("ReadinessChecker: failed to get event loop health: %s", e)
+        return result
+
+
+readiness_checker = ReadinessChecker()
--- a/backend/package/yuxi/channel/monitoring/state_aggregator.py
+++ b/backend/package/yuxi/channel/monitoring/state_aggregator.py
@ -0,0 +1,233 @@
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class ChannelHealthState:
+    channel_type: str
+    account_id: str
+    state: str = ""
+    connected: bool = False
+    healthy: bool = True
+    health_reason: str = "unknown"
+    health_detail: str | None = None
+    last_event_at: float = 0.0
+    last_error: str | None = None
+    active_runs: int = 0
+    uptime_seconds: float = 0.0
+    restart_count: int = 0
+
+
+@dataclass
+class ConfigHealthState:
+    status: str = "healthy"
+    last_known_good: str | None = None
+    issues: list[str] = field(default_factory=list)
+    revision: int = 0
+
+
+@dataclass
+class MonitoringState:
+    checks_total: int = 0
+    current_status: str = "unknown"
+    unhealthy_channels: list[str] = field(default_factory=list)
+    degraded_channels: list[str] = field(default_factory=list)
+    last_check_at: float = 0.0
+
+
+@dataclass
+class ReadinessState:
+    ready: bool = False
+    failing: list[str] = field(default_factory=list)
+    uptime_ms: float = 0.0
+    event_loop_degraded: bool = False
+
+
+@dataclass
+class DiagnosticState:
+    overall_status: str = "pass"
+    timestamp: float = 0.0
+    warnings_count: int = 0
+    security_warnings_count: int = 0
+
+
+@dataclass
+class GatewayConnectionState:
+    active_connections: int = 0
+    shutting_down: bool = False
+
+
+@dataclass
+class UnifiedState:
+    gateway_running: bool = False
+    gateway_started_at: float = 0.0
+    channels: dict[str, ChannelHealthState] = field(default_factory=dict)
+    config: ConfigHealthState = field(default_factory=ConfigHealthState)
+    monitoring: MonitoringState = field(default_factory=MonitoringState)
+    readiness: ReadinessState = field(default_factory=ReadinessState)
+    diagnostic: DiagnosticState = field(default_factory=DiagnosticState)
+    connections: GatewayConnectionState = field(default_factory=GatewayConnectionState)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "gateway": {
+                "running": self.gateway_running,
+                "started_at": self.gateway_started_at,
+            },
+            "channels": {
+                key: {
+                    "channel_type": ch.channel_type,
+                    "account_id": ch.account_id,
+                    "state": ch.state,
+                    "connected": ch.connected,
+                    "healthy": ch.healthy,
+                    "health_reason": ch.health_reason,
+                    "health_detail": ch.health_detail,
+                    "last_error": ch.last_error,
+                    "active_runs": ch.active_runs,
+                    "uptime_seconds": ch.uptime_seconds,
+                    "restart_count": ch.restart_count,
+                }
+                for key, ch in self.channels.items()
+            },
+            "config": {
+                "status": self.config.status,
+                "last_known_good": self.config.last_known_good,
+                "issues": self.config.issues,
+                "revision": self.config.revision,
+            },
+            "monitoring": {
+                "checks_total": self.monitoring.checks_total,
+                "current_status": self.monitoring.current_status,
+                "unhealthy_channels": self.monitoring.unhealthy_channels,
+                "degraded_channels": self.monitoring.degraded_channels,
+                "last_check_at": self.monitoring.last_check_at,
+            },
+            "readiness": {
+                "ready": self.readiness.ready,
+                "failing": self.readiness.failing,
+                "uptime_ms": self.readiness.uptime_ms,
+                "event_loop_degraded": self.readiness.event_loop_degraded,
+            },
+            "diagnostic": {
+                "overall_status": self.diagnostic.overall_status,
+                "warnings_count": self.diagnostic.warnings_count,
+                "security_warnings_count": self.diagnostic.security_warnings_count,
+            },
+            "connections": {
+                "active_connections": self.connections.active_connections,
+                "shutting_down": self.connections.shutting_down,
+            },
+        }
+
+    @property
+    def is_healthy(self) -> bool:
+        return (
+            self.readiness.ready
+            and self.monitoring.current_status == "healthy"
+            and not self.connections.shutting_down
+        )
+
+
+class StateAggregator:
+    def get_unified_state(self) -> UnifiedState:
+        state = UnifiedState()
+
+        self._collect_gateway_state(state)
+        self._collect_channel_health_state(state)
+        self._collect_config_health_state(state)
+        self._collect_monitoring_state(state)
+        self._collect_readiness_state(state)
+        self._collect_connection_state(state)
+
+        return state
+
+    def _collect_gateway_state(self, state: UnifiedState) -> None:
+        try:
+            from yuxi.channel.runtime.manager import gateway
+
+            state.gateway_running = gateway.is_running()
+            state.gateway_started_at = gateway._started_at
+        except Exception:
+            pass
+
+    def _collect_channel_health_state(self, state: UnifiedState) -> None:
+        try:
+            from yuxi.channel.monitoring.health_monitor import channel_health_monitor
+
+            metrics = channel_health_monitor.metrics
+            eval_details = metrics.evaluation_details
+
+            from yuxi.channel.runtime.manager import gateway
+
+            for key, snap in gateway.get_state().channels.items():
+                parts = key.split(":", 1)
+                if len(parts) != 2:
+                    continue
+                detail = eval_details.get(key, {})
+                state.channels[key] = ChannelHealthState(
+                    channel_type=parts[0],
+                    account_id=parts[1],
+                    state=snap.state.value,
+                    connected=snap.connected,
+                    healthy=detail.get("healthy", True),
+                    health_reason=detail.get("reason", "unknown"),
+                    health_detail=detail.get("detail"),
+                    last_event_at=snap.last_event_at,
+                    last_error=snap.last_error,
+                    active_runs=snap.active_runs,
+                    uptime_seconds=snap.uptime_seconds,
+                    restart_count=snap.restart_count,
+                )
+        except Exception:
+            pass
+
+    def _collect_config_health_state(self, state: UnifiedState) -> None:
+        try:
+            pass
+        except Exception:
+            pass
+
+    def _collect_monitoring_state(self, state: UnifiedState) -> None:
+        try:
+            from yuxi.channel.monitoring.health_monitor import channel_health_monitor
+
+            metrics = channel_health_monitor.metrics
+            state.monitoring = MonitoringState(
+                checks_total=metrics.checks_total,
+                current_status=metrics.current_status,
+                unhealthy_channels=list(metrics.unhealthy_channels),
+                degraded_channels=list(metrics.degraded_channels),
+                last_check_at=metrics.last_check_at,
+            )
+        except Exception:
+            pass
+
+    def _collect_readiness_state(self, state: UnifiedState) -> None:
+        try:
+            from yuxi.channel.monitoring.readiness import readiness_checker
+
+            last = readiness_checker._last_result
+            if last is not None:
+                state.readiness = ReadinessState(
+                    ready=last.ready,
+                    failing=list(last.failing),
+                    uptime_ms=last.uptime_ms,
+                    event_loop_degraded=last.event_loop_degraded,
+                )
+        except Exception:
+            pass
+
+    def _collect_connection_state(self, state: UnifiedState) -> None:
+        try:
+            from yuxi.channel.gateway.server import gateway_ws_server
+
+            state.connections = GatewayConnectionState(
+                active_connections=gateway_ws_server.active_count,
+                shutting_down=gateway_ws_server.shutting_down,
+            )
+        except Exception:
+            pass
+
+
+state_aggregator = StateAggregator()