feat(servers): 主列表宝塔登录 + 健康检查写心跳

主服务器列表操作列一键登录宝塔；SSH 健康检查写入 Redis 心跳并刷新在线状态。

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_server_connectivity.py

@@ -9,6 +9,8 @@ import pytest
 from server.application.server_connectivity import (
     count_monitored_connectivity,
     item_matches_online_filter,
+    normalize_ssh_probe_system_info,
+    write_ssh_health_heartbeat,
 )
 
 
@@ -42,3 +44,50 @@ async def test_count_monitored_connectivity_batches_redis():
     out = await count_monitored_connectivity(redis, session, batch_size=2)
 
     assert out == {"monitored": 3, "online": 1, "offline": 2}
+
+
+def test_normalize_ssh_probe_system_info_nested():
+    probe = {
+        "status": "online",
+        "channel": "ssh",
+        "system_info": {
+            "status": "healthy",
+            "system_info": {"cpu_usage": 12.5, "probe": "ssh"},
+        },
+    }
+    out = normalize_ssh_probe_system_info(probe)
+    assert out["cpu_usage"] == 12.5
+    assert out["probe_channel"] == "ssh"
+
+
+def test_normalize_ssh_probe_system_info_offline_error():
+    probe = {"status": "offline", "channel": "ssh", "error": "timeout"}
+    out = normalize_ssh_probe_system_info(probe)
+    assert out["probe_error"] == "timeout"
+    assert out["probe_channel"] == "ssh"
+
+
+@pytest.mark.asyncio
+async def test_write_ssh_health_heartbeat_redis_shape():
+    redis = AsyncMock()
+    server = MagicMock()
+    server.id = 42
+    server.agent_version = "2.1.0"
+
+    ts = await write_ssh_health_heartbeat(
+        redis,
+        server,
+        {
+            "status": "online",
+            "channel": "ssh",
+            "system_info": {"status": "healthy", "system_info": {"cpu_usage": 1.0}},
+        },
+    )
+
+    assert ts
+    redis.hset.assert_awaited_once()
+    mapping = redis.hset.await_args.kwargs["mapping"]
+    assert mapping["is_online"] == "True"
+    assert mapping["agent_version"] == "2.1.0"
+    assert "cpu_usage" in mapping["system_info"]
+    redis.expire.assert_awaited_once()