feat: 数据面故障韧性 — 节点健康上报(控制面) + 客户端连通看门狗

盲区:节点 status 只看 agent 的 gRPC 在线,sing-box 数据面坏了(崩/配置坏/数据口不通)
而 agent 仍在线时,节点仍显示 up、connect 放行、客户端「已连接」但流量全失败,且不自愈。

控制面:
- agent 每心跳探 sing-box clash_api /version(2s 超时,连续 2 次失败才报不健康,防抖),
  经新增 HeartbeatRequest.data_plane_healthy 上报(手写 agentv1 契约 + proto 同步;JSON codec)。
- NodeLoad 加 DataPlaneHealthy(随 load 写 Redis;字段缺失默认 healthy 防滚动期误杀)。
- effectiveNodeStatus 扩为 (dbStatus, agentOnline, dataPlaneHealthy);ListNodes 与 ConnectNode
  统一改用它 → 数据面坏的节点列表置灰 + connect 返回 404 拦截。

客户端(connection_provider 连通看门狗):
- 连上后每 15s 经隧道 HTTP 探海外 generate_204(可注入),连续 3 次失败判当前节点不可用。
- 智能选择 → 自动切到其他最优可用节点重连;手动选定 → 断开并提示「节点异常」(尊重用户选择)。
- 新增 nodeUnhealthySwitched/nodeUnhealthyError 文案。

测试:agent 健康探测防抖、effectiveNodeStatus 真值表、dataPlaneHealthy 助手、看门狗
智能切/手动断/健康不触发;go test ./... 与 flutter test 全绿。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

client/lib/l10n/app_text.dart

@@ -56,6 +56,9 @@ abstract class AppText {
   String nodeSwitchTitle(String name);
   String get nodeSwitchBody;
   String get nodeSwitchConfirm;
+  // 连通看门狗:当前节点数据面不可用时的提示
+  String get nodeUnhealthySwitched; // 智能选择已自动切换
+  String get nodeUnhealthyError; // 手动节点:断开并提示
   String get smartSub;
   String get recommended;
 

client/lib/l10n/strings_en.dart

@@ -70,6 +70,10 @@ class StringsEn extends AppText {
   @override
   String get nodeSwitchConfirm => 'Switch';
   @override
+  String get nodeUnhealthySwitched => 'Current node unreachable — switched automatically';
+  @override
+  String get nodeUnhealthyError => 'Current node is unreachable — reconnect or switch node';
+  @override
   String get smartSub => 'Picks the best node for your network';
   @override
   String get recommended => 'Recommended';

client/lib/l10n/strings_zh.dart

@@ -69,6 +69,10 @@ class StringsZh extends AppText {
   @override
   String get nodeSwitchConfirm => '切换';
   @override
+  String get nodeUnhealthySwitched => '当前节点异常，已自动切换';
+  @override
+  String get nodeUnhealthyError => '当前节点异常，请重连或更换节点';
+  @override
   String get smartSub => '根据当前网络环境,自动选择最优节点';
   @override
   String get recommended => '推荐';

client/lib/state/connection_provider.dart

@@ -6,12 +6,14 @@
 //   - 严禁乐观翻转：VpnPhase.on 必须由 bridge.statusStream 确认后才置。
 //   - 状态来源：bridge.statusStream（来自内核回调），非 Timer 模拟。
 import 'dart:async';
+import 'dart:io';
 
 import 'package:flutter_riverpod/flutter_riverpod.dart';
 
 import '../bridge/vpn_bridge.dart';
 import '../bridge/vpn_bridge_provider.dart';
 import '../l10n/app_text.dart';
+import '../models/node.dart';
 import '../services/api_config.dart';
 import '../services/connect_api.dart';
 import '../services/device_identity.dart';
@@ -53,19 +55,59 @@ class ConnectionState {
   int get hashCode => Object.hash(phase, elapsed, error);
 }
 
+// ── 连通看门狗 ─────────────────────────────────────────────────────
+//
+// 已连接(本地 TUN 已起)≠ 远端节点数据面可用:节点 sing-box 崩了/数据口不通而
+// agent 仍在线时,客户端会一直显示「已连接」但流量全失败。看门狗周期性经隧道做
+// 连通性探测,连续失败即判当前节点不可用 → 智能选择自动切节点 / 手动选定的只断开提示。
+
+/// 连通探测接缝:true=经隧道能到外网。默认走海外 204 端点;测试可注入假实现。
+typedef HealthProber = Future<bool> Function();
+
+const _kWatchdogInterval = Duration(seconds: 15);
+const _kHealthFailLimit = 3; // 连续 N 次失败(≈45s)才判定不可用,防抖
+
+/// 默认探测:GET 海外 generate_204(全局 TUN 下走隧道 → 真实检验代理出网)。
+/// 任一端点 2xx/204 即视为通;多端点容错,避免单点误判。
+Future<bool> defaultHealthProbe() async {
+  const urls = [
+    'https://www.gstatic.com/generate_204',
+    'https://cp.cloudflare.com/generate_204',
+  ];
+  for (final u in urls) {
+    final client = HttpClient()..connectionTimeout = const Duration(seconds: 4);
+    try {
+      final req = await client.getUrl(Uri.parse(u)).timeout(const Duration(seconds: 5));
+      final resp = await req.close().timeout(const Duration(seconds: 5));
+      await resp.drain<void>();
+      if (resp.statusCode >= 200 && resp.statusCode < 300) return true;
+    } catch (_) {
+      // 该端点不通,试下一个
+    } finally {
+      client.close(force: true);
+    }
+  }
+  return false;
+}
+
 // ── 状态机 ───────────────────────────────────────────────────────
 
 class ConnectionController extends StateNotifier<ConnectionState> {
-  ConnectionController(this._ref, this._bridge)
-      : super(const ConnectionState(phase: VpnPhase.off)) {
+  ConnectionController(this._ref, this._bridge, {HealthProber? prober})
+      : _prober = prober ?? defaultHealthProbe,
+        super(const ConnectionState(phase: VpnPhase.off)) {
     // 订阅桥状态流：状态由内核事件驱动，严禁 UI 乐观翻转。
     _statusSub = _bridge.statusStream.listen(_onKernelStatus);
   }
 
   final Ref _ref;
   final VpnBridge _bridge;
+  final HealthProber _prober;
   StreamSubscription<VpnStatus>? _statusSub;
   Timer? _elapsed;
+  Timer? _watchdog;
+  int _healthFails = 0;
+  bool _probing = false;
   ConnectApi? _api;
 
   // ── 公有 API ───────────────────────────────────────────────────
@@ -167,16 +209,84 @@ class ConnectionController extends StateNotifier<ConnectionState> {
       case VpnStatus.on:
         state = state.copyWith(phase: VpnPhase.on);
         _startElapsed();
+        _startWatchdog();
       case VpnStatus.connecting:
         state = state.copyWith(phase: VpnPhase.connecting);
         _stopElapsed();
+        _stopWatchdog();
       case VpnStatus.off:
       case VpnStatus.error:
         state = const ConnectionState(phase: VpnPhase.off);
         _stopElapsed();
+        _stopWatchdog();
     }
   }
 
+  // ── 连通看门狗 ───────────────────────────────────────────────
+  void _startWatchdog() {
+    _watchdog?.cancel();
+    _healthFails = 0;
+    _watchdog = Timer.periodic(_kWatchdogInterval, (_) => _checkHealth());
+  }
+
+  void _stopWatchdog() {
+    _watchdog?.cancel();
+    _watchdog = null;
+    _healthFails = 0;
+    _probing = false;
+  }
+
+  Future<void> _checkHealth() async {
+    if (_probing || !mounted || state.phase != VpnPhase.on) return;
+    _probing = true;
+    final ok = await _prober();
+    _probing = false;
+    if (!mounted || state.phase != VpnPhase.on) return;
+    if (ok) {
+      _healthFails = 0;
+      return;
+    }
+    _healthFails++;
+    if (_healthFails >= _kHealthFailLimit) {
+      _stopWatchdog();
+      await _onNodeUnhealthy();
+    }
+  }
+
+  /// 当前节点连续探测失败:智能选择 → 切到其他最优可用节点重连;手动选定 → 断开并提示。
+  Future<void> _onNodeUnhealthy() async {
+    final t = _ref.read(appTextProvider);
+    final smart = _ref.read(selectedNodeCodeProvider) == kSmartNodeCode;
+    final altCode = smart ? _pickAlternativeCode(_ref.read(effectiveNodeProvider).code) : null;
+    if (smart && altCode != null) {
+      _ref.read(selectedNodeCodeProvider.notifier).state = altCode;
+      await _disconnect();
+      await _connect();
+      // 重连进行中给出「已自动切换」提示(connecting 态显示,连上后随 copyWith 清除)。
+      if (mounted && state.phase != VpnPhase.off) {
+        state = ConnectionState(phase: state.phase, error: t.nodeUnhealthySwitched);
+      }
+      return;
+    }
+    // 手动选定节点(或智能模式无其他可用节点):断开并提示,尊重用户选择、不自动换。
+    await _disconnect();
+    if (mounted) state = ConnectionState(phase: VpnPhase.off, error: t.nodeUnhealthyError);
+  }
+
+  /// 选延迟最优、可用(status up)、非当前节点的 code;无则 null。
+  String? _pickAlternativeCode(String excludeCode) {
+    final nodes = (_ref.read(nodesProvider).valueOrNull ?? const [])
+        .where((n) => !n.isDown && n.uuid.isNotEmpty && n.code != excludeCode)
+        .toList();
+    if (nodes.isEmpty) return null;
+    nodes.sort((a, b) {
+      final pa = a.ping > 0 ? a.ping : 1 << 30;
+      final pb = b.ping > 0 ? b.ping : 1 << 30;
+      return pa.compareTo(pb);
+    });
+    return nodes.first.code;
+  }
+
   void _startElapsed() {
     _elapsed?.cancel();
     _elapsed = Timer.periodic(const Duration(seconds: 1), (_) {
@@ -195,6 +305,7 @@ class ConnectionController extends StateNotifier<ConnectionState> {
   void dispose() {
     _statusSub?.cancel();
     _stopElapsed();
+    _stopWatchdog();
     _api?.dispose();
     super.dispose();
   }

client/test/unit/connection_watchdog_test.dart

@@ -0,0 +1,132 @@
+// connection_watchdog_test.dart — 连通看门狗:已连接后探测连续失败时,
+// 智能选择自动切到其他节点;手动选定的节点只断开并提示。
+import 'dart:async';
+
+import 'package:flutter/widgets.dart';
+import 'package:flutter_riverpod/flutter_riverpod.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:pangolin_vpn/bridge/vpn_bridge.dart';
+import 'package:pangolin_vpn/bridge/vpn_bridge_provider.dart';
+import 'package:pangolin_vpn/l10n/strings_zh.dart';
+import 'package:pangolin_vpn/models/node.dart';
+import 'package:pangolin_vpn/services/connect_api.dart';
+import 'package:pangolin_vpn/state/connection_provider.dart';
+import 'package:pangolin_vpn/state/nodes_provider.dart';
+
+// 可控假桥:能手动推 VpnStatus,无内部计时器(避免 pending timer)。
+class _FakeBridge implements VpnBridge {
+  final _status = StreamController<VpnStatus>.broadcast();
+  final _stats = StreamController<VpnStatsEvent>.broadcast();
+  void emit(VpnStatus s) => _status.add(s);
+  @override
+  Stream<VpnStatus> get statusStream => _status.stream;
+  @override
+  Stream<VpnStatsEvent> get statsStream => _stats.stream;
+  @override
+  Future<void> start(String configJson) async {}
+  @override
+  Future<void> stop() async {}
+  @override
+  Future<VpnStatus> getStatus() async => VpnStatus.off;
+  @override
+  Future<void> selectOutbound(String tag) async {}
+  @override
+  Future<String> getActiveOutbound() async => 'auto';
+  @override
+  Future<void> setKillSwitch({required bool on}) async {}
+  @override
+  void dispose() {
+    _status.close();
+    _stats.close();
+  }
+}
+
+// 重连时 fetchConfig 立即抛错,绕开真实网络(不留超时计时器)。
+class _FakeConnectApi extends ConnectApi {
+  _FakeConnectApi() : super(baseUrl: 'http://test.local', authToken: '');
+  @override
+  Future<String> fetchConfig({required String nodeId, required String deviceId, bool splitCN = false}) async {
+    throw const ConnectApiException(statusCode: 0, messageZh: 'x', messageEn: 'x');
+  }
+  @override
+  void dispose() {}
+}
+
+class _StubNodes extends NodesNotifier {
+  @override
+  Future<List<Node>> build() async => const [
+        Node(code: 'HK', nameZh: '香港', nameEn: 'Hong Kong', ping: 20, uuid: 'hk-uuid', host: 'hk', port: 443),
+        Node(code: 'JP', nameZh: '东京', nameEn: 'Tokyo', ping: 35, uuid: 'jp-uuid', host: 'jp', port: 443),
+      ];
+  @override
+  Future<void> refresh() async {}
+}
+
+void main() {
+  const t = StringsZh();
+
+  ProviderContainer makeContainer(_FakeBridge bridge, {required bool healthy}) =>
+      ProviderContainer(overrides: [
+        vpnBridgeProvider.overrideWithValue(bridge),
+        nodesProvider.overrideWith(_StubNodes.new),
+        connectApiFactoryProvider.overrideWithValue((_) => _FakeConnectApi()),
+        connectionProvider.overrideWith((ref) => ConnectionController(
+              ref, ref.watch(vpnBridgeProvider),
+              prober: () async => healthy,
+            )),
+      ]);
+
+  // 驱动到 on,再走 3 个看门狗周期(每个 15s)。
+  Future<void> driveOnAndProbe(WidgetTester tester, ProviderContainer c, _FakeBridge bridge) async {
+    c.read(nodesProvider); // 触发(异步)节点加载
+    await tester.pump(); // 等 valueOrNull 就绪(智能切换要读节点列表)
+    c.read(connectionProvider); // 建 controller(订阅状态流)
+    bridge.emit(VpnStatus.on);
+    await tester.pump();
+    expect(c.read(connectionProvider).phase, VpnPhase.on);
+    for (var i = 0; i < 3; i++) {
+      await tester.pump(const Duration(seconds: 15));
+      await tester.pump();
+    }
+    await tester.pump();
+  }
+
+  testWidgets('智能模式:探测连续失败 → 自动切到其他最优节点', (tester) async {
+    final bridge = _FakeBridge();
+    final c = makeContainer(bridge, healthy: false);
+    addTearDown(bridge.dispose);
+    addTearDown(c.dispose);
+    await tester.pumpWidget(UncontrolledProviderScope(container: c, child: const SizedBox()));
+    // 默认即智能选择(selectedNodeCode = AUTO);有效节点取最小延迟 HK,故应切到 JP。
+    await driveOnAndProbe(tester, c, bridge);
+    expect(c.read(selectedNodeCodeProvider), 'JP', reason: '智能模式应自动切到其他最优节点');
+  });
+
+  testWidgets('手动选定节点:探测连续失败 → 断开并提示,不自动换', (tester) async {
+    final bridge = _FakeBridge();
+    final c = makeContainer(bridge, healthy: false);
+    addTearDown(bridge.dispose);
+    addTearDown(c.dispose);
+    await tester.pumpWidget(UncontrolledProviderScope(container: c, child: const SizedBox()));
+    c.read(selectedNodeCodeProvider.notifier).state = 'HK'; // 手动选定 HK
+    await driveOnAndProbe(tester, c, bridge);
+    expect(c.read(selectedNodeCodeProvider), 'HK', reason: '手动模式不应自动换节点');
+    final st = c.read(connectionProvider);
+    expect(st.phase, VpnPhase.off, reason: '应断开');
+    expect(st.error, t.nodeUnhealthyError, reason: '应给出节点异常提示');
+  });
+
+  testWidgets('探测正常:不触发任何动作', (tester) async {
+    final bridge = _FakeBridge();
+    final c = makeContainer(bridge, healthy: true);
+    addTearDown(bridge.dispose);
+    addTearDown(c.dispose);
+    await tester.pumpWidget(UncontrolledProviderScope(container: c, child: const SizedBox()));
+    c.read(selectedNodeCodeProvider.notifier).state = 'HK';
+    await driveOnAndProbe(tester, c, bridge);
+    expect(c.read(connectionProvider).phase, VpnPhase.on, reason: '健康时应保持连接');
+    expect(c.read(selectedNodeCodeProvider), 'HK');
+    bridge.emit(VpnStatus.off); // 收尾:停看门狗/计时器,避免 pending timer
+    await tester.pump();
+  });
+}

server/cmd/server/main.go

@@ -322,7 +322,7 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv
 			slog.Warn("PANGOLIN_PUBLIC_URL 未设置:国内分流(split_cn)将被静默跳过,客户端全量走隧道。" +
 				"如需国内直连,设为控制面对外公网基址(如 http://<公网IP>:8080)")
 		}
-		nodeAPI = httpapi.NewNodeAPI(nodeStore, nodeSvc.Hub(), os.Getenv("NODE_DERIVE_KEY"), publicURL)
+		nodeAPI = httpapi.NewNodeAPI(nodeStore, nodeSvc.Hub(), nodeSvc.Load(), os.Getenv("NODE_DERIVE_KEY"), publicURL)
 	}
 
 	// 国内分流(#5)的 rule-set 静态服务:GET /v1/rules/{name}.srs(自托管,

server/internal/agentd/agent.go

@@ -23,6 +23,7 @@ type Agent struct {
 	enrollDial EnrollDialer
 	load       LoadSource
 	usage      UsageSource
+	health     HealthSource
 	clock      func() time.Time
 
 	lastCommandID atomic.Int64
@@ -39,6 +40,7 @@ func WithDialer(d Dialer) Option             { return func(a *Agent) { a.dial =
 func WithEnrollDialer(d EnrollDialer) Option { return func(a *Agent) { a.enrollDial = d } }
 func WithLoadSource(l LoadSource) Option     { return func(a *Agent) { a.load = l } }
 func WithUsageSource(u UsageSource) Option   { return func(a *Agent) { a.usage = u } }
+func WithHealthSource(h HealthSource) Option  { return func(a *Agent) { a.health = h } }
 func WithClock(f func() time.Time) Option    { return func(a *Agent) { a.clock = f } }
 
 // New builds an Agent. Production callers pass nothing extra and get mTLS dialers;
@@ -71,6 +73,9 @@ func New(cfg Config, opts ...Option) *Agent {
 	if a.usage == nil {
 		a.usage = nopUsageSource{}
 	}
+	if a.health == nil {
+		a.health = newClashHealthSource()
+	}
 	return a
 }
 

server/internal/agentd/heartbeat.go

@@ -2,6 +2,7 @@ package agentd
 
 import (
 	"context"
+	"net/http"
 	"time"
 
 	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
@@ -23,6 +24,49 @@ func (d defaultLoadSource) Load() (int32, int64, int64, float64) {
 	return int32(d.sb.OnlinePeers()), 0, 0, 0
 }
 
+// HealthSource reports whether the local data plane (sing-box) is healthy, sent
+// in each heartbeat so the control plane can mark a node down when its sing-box
+// is dead even though the agent's gRPC stream is still up.
+type HealthSource interface{ Healthy() bool }
+
+// clashHealthSource probes sing-box's clash_api /version. A reachable, 200-OK API
+// means the sing-box process is alive and its config loaded. Applies a 2-strike
+// debounce: a single failed probe is tolerated (covers the sing-box restart
+// window); two consecutive failures report unhealthy.
+type clashHealthSource struct {
+	probe func() bool // single probe, true = healthy; injectable for tests
+	fails int
+}
+
+func newClashHealthSource() *clashHealthSource {
+	client := &http.Client{Timeout: 2 * time.Second}
+	return &clashHealthSource{probe: func() bool {
+		req, err := http.NewRequestWithContext(
+			context.Background(), http.MethodGet, "http://"+clashAPIAddr+"/version", nil)
+		if err != nil {
+			return false
+		}
+		if clashAPISecret != "" {
+			req.Header.Set("Authorization", "Bearer "+clashAPISecret)
+		}
+		resp, err := client.Do(req)
+		if err != nil {
+			return false
+		}
+		defer resp.Body.Close()
+		return resp.StatusCode == http.StatusOK
+	}}
+}
+
+func (h *clashHealthSource) Healthy() bool {
+	if h.probe() {
+		h.fails = 0
+		return true
+	}
+	h.fails++
+	return h.fails < 2 // tolerate 1 failure; unhealthy from the 2nd consecutive
+}
+
 // runHeartbeat sends a heartbeat every cfg.HeartbeatInterval until ctx is
 // cancelled or an RPC error occurs. A need_full_resync response returns
 // errResync so the session re-Registers and overwrites local state.
@@ -51,6 +95,7 @@ func (a *Agent) sendHeartbeat(ctx context.Context, client agentv1.AgentServiceCl
 		BandwidthDownBps: down,
 		CPUPercent:       cpu,
 		TimestampUnix:    a.clock().Unix(),
+		DataPlaneHealthy: a.health.Healthy(),
 	})
 	if err != nil {
 		return err

server/internal/agentd/heartbeat_health_test.go

@@ -0,0 +1,23 @@
+package agentd
+
+import "testing"
+
+// TestClashHealthSource_Debounce verifies the 2-strike debounce: a single failed
+// probe is tolerated (sing-box restart window), two consecutive failures report
+// unhealthy, and a success resets the strike counter.
+func TestClashHealthSource_Debounce(t *testing.T) {
+	probeResults := []bool{true, false, false, false, true, false}
+	want := []bool{true, true, false, false, true, true}
+	//             ↑ok  ↑1st ↑2nd  ↑3rd  ↑ok  ↑1st-after-reset
+	i := 0
+	h := &clashHealthSource{probe: func() bool {
+		r := probeResults[i]
+		i++
+		return r
+	}}
+	for n, w := range want {
+		if got := h.Healthy(); got != w {
+			t.Errorf("Healthy() call %d = %v, want %v (probe=%v)", n, got, w, probeResults[n])
+		}
+	}
+}

server/internal/httpapi/nodes.go

@@ -1,6 +1,7 @@
 package httpapi
 
 import (
+	"context"
 	"encoding/json"
 	"log/slog"
 	"net/http"
@@ -23,19 +24,40 @@ const (
 	freeMinuteTTL = time.Minute
 )
 
+// nodeLoadReader reads a node's last-reported runtime load (for the data-plane
+// health flag). *nodes.LoadCache satisfies it; nil = treat all nodes healthy.
+type nodeLoadReader interface {
+	Get(ctx context.Context, nodeUUID string) (*nodes.NodeLoad, bool, error)
+}
+
 // NodeAPI serves the /v1/nodes endpoints.
 type NodeAPI struct {
 	store     nodes.NodeStore
 	hub       *nodes.Hub
+	load      nodeLoadReader
 	deriveKey string
 	// rulesBaseURL 是控制面对外公网基址(PANGOLIN_PUBLIC_URL),供国内分流的
 	// rule_set .srs 下载用;空则分流不生效。
 	rulesBaseURL string
 }
 
-// NewNodeAPI creates a NodeAPI.
-func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, deriveKey, rulesBaseURL string) *NodeAPI {
-	return &NodeAPI{store: store, hub: hub, deriveKey: deriveKey, rulesBaseURL: rulesBaseURL}
+// NewNodeAPI creates a NodeAPI. load may be nil (then all nodes are treated as
+// data-plane healthy — agent gRPC liveness still gates status).
+func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, load nodeLoadReader, deriveKey, rulesBaseURL string) *NodeAPI {
+	return &NodeAPI{store: store, hub: hub, load: load, deriveKey: deriveKey, rulesBaseURL: rulesBaseURL}
+}
+
+// dataPlaneHealthy reports the node's last sing-box health (default true when
+// unknown: just registered / load expired / read error — agent liveness gates).
+func (a *NodeAPI) dataPlaneHealthy(ctx context.Context, nodeUUID string) bool {
+	if a.load == nil {
+		return true
+	}
+	l, ok, err := a.load.Get(ctx, nodeUUID)
+	if err != nil || !ok || l == nil {
+		return true
+	}
+	return l.DataPlaneHealthy
 }
 
 // ─── GET /v1/nodes ───────────────────────────────────────────────────────────
@@ -53,12 +75,13 @@ type nodeResponse struct {
 }
 
 // effectiveNodeStatus downgrades a DB-'up' node to "down" when its agent is
-// offline. The DB status column reflects provisioning/scheduler state, NOT agent
-// liveness — without this, a node whose agent has been offline for days still
-// shows healthy (the gap that hid a 6-day agent outage and let clients "connect"
-// to a node that couldn't serve them).
-func effectiveNodeStatus(dbStatus string, agentOnline bool) string {
-	if dbStatus == "up" && !agentOnline {
+// offline OR its data plane (sing-box) is unhealthy. The DB status column reflects
+// provisioning/scheduler state, NOT live health — without this, a node whose agent
+// has been offline for days, or whose sing-box has crashed while the agent stays
+// connected, still shows healthy and lets clients "connect" to a node that can't
+// serve them.
+func effectiveNodeStatus(dbStatus string, agentOnline, dataPlaneHealthy bool) string {
+	if dbStatus == "up" && (!agentOnline || !dataPlaneHealthy) {
 		return "down"
 	}
 	return dbStatus
@@ -85,7 +108,9 @@ func (a *NodeAPI) ListNodes(w http.ResponseWriter, r *http.Request) {
 			NameZH: n.NameZH,
 			NameEN: n.NameEN,
 			Tier:   n.Tier,
-			Status: effectiveNodeStatus(n.Status, a.hub == nil || a.hub.IsOnline(n.UUID)),
+			Status: effectiveNodeStatus(n.Status,
+				a.hub == nil || a.hub.IsOnline(n.UUID),
+				a.dataPlaneHealthy(r.Context(), n.UUID)),
 			Host:   host,
 			Port:   port,
 		})
@@ -188,7 +213,9 @@ func (a *NodeAPI) ConnectNode(w http.ResponseWriter, r *http.Request) {
 		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
 		return
 	}
-	if node == nil || node.Status != "up" {
+	if node == nil || effectiveNodeStatus(node.Status,
+		a.hub == nil || a.hub.IsOnline(nodeUUID),
+		a.dataPlaneHealthy(r.Context(), nodeUUID)) != "up" {
 		apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
 		return
 	}

server/internal/httpapi/nodes_status_test.go

@@ -1,25 +1,69 @@
 package httpapi
 
-import "testing"
+import (
+	"context"
+	"errors"
+	"testing"
 
-// TestEffectiveNodeStatus guards the "don't show a node healthy when its agent is
-// offline" fix: a DB-'up' node with an offline agent must report "down".
-func TestEffectiveNodeStatus(t *testing.T) {
+	"github.com/wangjia/pangolin/server/internal/nodes"
+)
+
+type fakeLoadReader struct {
+	load *nodes.NodeLoad
+	ok   bool
+	err  error
+}
+
+func (f fakeLoadReader) Get(context.Context, string) (*nodes.NodeLoad, bool, error) {
+	return f.load, f.ok, f.err
+}
+
+// TestDataPlaneHealthy: unknown (nil reader / missing / error) defaults to healthy
+// (agent liveness gates); a present load returns its reported flag.
+func TestDataPlaneHealthy(t *testing.T) {
+	ctx := context.Background()
 	cases := []struct {
-		name   string
-		db     string
-		online bool
-		want   string
+		name string
+		api  *NodeAPI
+		want bool
 	}{
-		{"up + agent online", "up", true, "up"},
-		{"up + agent offline → down", "up", false, "down"}, // the core fix
-		{"draining untouched when offline", "draining", false, "draining"},
-		{"down stays down", "down", true, "down"},
+		{"nil reader → healthy", &NodeAPI{}, true},
+		{"missing load → healthy", &NodeAPI{load: fakeLoadReader{ok: false}}, true},
+		{"read error → healthy", &NodeAPI{load: fakeLoadReader{err: errors.New("boom")}}, true},
+		{"present healthy", &NodeAPI{load: fakeLoadReader{load: &nodes.NodeLoad{DataPlaneHealthy: true}, ok: true}}, true},
+		{"present unhealthy → false", &NodeAPI{load: fakeLoadReader{load: &nodes.NodeLoad{DataPlaneHealthy: false}, ok: true}}, false},
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			if got := effectiveNodeStatus(c.db, c.online); got != c.want {
-				t.Errorf("effectiveNodeStatus(%q, %v) = %q, want %q", c.db, c.online, got, c.want)
+			if got := c.api.dataPlaneHealthy(ctx, "n1"); got != c.want {
+				t.Errorf("dataPlaneHealthy = %v, want %v", got, c.want)
+			}
+		})
+	}
+}
+
+// TestEffectiveNodeStatus guards the "don't show a node healthy when it can't
+// serve" fix: a DB-'up' node reports "down" when its agent is offline OR its data
+// plane (sing-box) is unhealthy.
+func TestEffectiveNodeStatus(t *testing.T) {
+	cases := []struct {
+		name    string
+		db      string
+		online  bool
+		healthy bool
+		want    string
+	}{
+		{"up + agent online + healthy", "up", true, true, "up"},
+		{"up + agent offline → down", "up", false, true, "down"},          // agent-liveness fix
+		{"up + agent online + dp unhealthy → down", "up", true, false, "down"}, // data-plane fix
+		{"up + offline + unhealthy → down", "up", false, false, "down"},
+		{"draining untouched", "draining", false, false, "draining"},
+		{"down stays down", "down", true, true, "down"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			if got := effectiveNodeStatus(c.db, c.online, c.healthy); got != c.want {
+				t.Errorf("effectiveNodeStatus(%q, %v, %v) = %q, want %q", c.db, c.online, c.healthy, got, c.want)
 			}
 		})
 	}

server/internal/nodes/handler_grpc.go

@@ -183,6 +183,7 @@ func (h *Handler) Heartbeat(ctx context.Context, req *agentv1.HeartbeatRequest)
 		BandwidthUpBps:   req.BandwidthUpBps,
 		BandwidthDownBps: req.BandwidthDownBps,
 		CPUPercent:       req.CPUPercent,
+		DataPlaneHealthy: req.DataPlaneHealthy,
 	}); err != nil {
 		slog.Warn("nodes.Handler.Heartbeat: load write failed",
 			"node_uuid", nodeUUID, "err", err)

server/internal/nodes/load.go

@@ -24,6 +24,10 @@ type NodeLoad struct {
 	BandwidthUpBps   int64
 	BandwidthDownBps int64
 	CPUPercent       float64
+	// DataPlaneHealthy is the agent's last sing-box health probe result. The
+	// control plane downgrades a node to "down" when this is false (even if the
+	// agent's gRPC stream is still up). Absent in older hashes → read as true.
+	DataPlaneHealthy bool
 }
 
 // LoadCache reads and writes node:load Redis hashes.
@@ -41,12 +45,17 @@ func NewLoadCache(rdb redis.Cmdable) *LoadCache {
 // The hash fields match the names in doc/03 §4: online_count, bw_up, bw_down, cpu.
 func (c *LoadCache) Set(ctx context.Context, nodeUUID string, load NodeLoad) error {
 	key := nodeLoadKeyPrefix + nodeUUID
+	dpHealthy := 0
+	if load.DataPlaneHealthy {
+		dpHealthy = 1
+	}
 	pipe := c.rdb.Pipeline()
 	pipe.HSet(ctx, key,
 		"online_count", load.OnlineCount,
 		"bw_up", load.BandwidthUpBps,
 		"bw_down", load.BandwidthDownBps,
 		"cpu", load.CPUPercent,
+		"dp_healthy", dpHealthy,
 	)
 	pipe.Expire(ctx, key, nodeLoadTTL)
 	if _, err := pipe.Exec(ctx); err != nil {
@@ -66,7 +75,12 @@ func (c *LoadCache) Get(ctx context.Context, nodeUUID string) (*NodeLoad, bool,
 	if len(vals) == 0 {
 		return nil, false, nil
 	}
-	var l NodeLoad
+	// Default healthy: hashes written before dp_healthy existed (or by an older
+	// control plane during rollout) must not be read as "unhealthy" → false-down.
+	l := NodeLoad{DataPlaneHealthy: true}
+	if v, ok := vals["dp_healthy"]; ok {
+		l.DataPlaneHealthy = v == "1"
+	}
 	if v, ok := vals["online_count"]; ok {
 		_, _ = fmt.Sscan(v, &l.OnlineCount)
 	}

server/internal/pb/agentv1/types.go

@@ -108,6 +108,9 @@ type HeartbeatRequest struct {
 	BandwidthDownBps int64   `json:"bandwidth_down_bps,omitempty"`
 	CPUPercent       float64 `json:"cpu_percent,omitempty"`
 	TimestampUnix    int64   `json:"timestamp_unix,omitempty"`
+	// DataPlaneHealthy: agent 本地探测 sing-box(clash_api) 是否健康。false → 控制面把
+	// 该节点判为 down(列表置灰 + connect 拦截),即便 agent 自身在线。
+	DataPlaneHealthy bool `json:"data_plane_healthy,omitempty"`
 }
 
 type HeartbeatResponse struct {

server/proto/agent/v1/agent.proto

@@ -127,6 +127,9 @@ message HeartbeatRequest {
   int64 bandwidth_down_bps = 5;
   double cpu_percent = 6;
   int64 timestamp_unix = 7;
+  // data_plane_healthy: agent 本地探测 sing-box(clash_api)是否健康。false → 控制面把
+  // 该节点判为 down(列表置灰 + connect 拦截),即便 agent 自身在线。
+  bool data_plane_healthy = 8;
 }
 
 message HeartbeatResponse {