fix: 六轮深度扫描 — 47项Bug修复、安全加固、死代码清理

Critical runtime bugs:
- terminal.html WebSSH完全不可用（URL前缀/JSON解析/Content-Type三处错误）
- servers.py路由遮蔽：/logs被/{id}拦截，3个前端页面同步日志查询失败
- scripts.html startExecPoll()→startExecPolling()，长任务快速执行崩溃
- agent.py {value!r!s:.50}格式串非法，agent发非数值时ValueError
- alerts.html d.daily.reduce()无null检查，API返回空数据时TypeError

Resource leak / stability:
- websocket.py僵尸连接未关闭TCP，文件描述符泄漏
- websocket.py _last_alert_time字典无限增长（加1小时过期清理）
- asyncssh_pool.py全忙时超过MAX_CONNECTIONS无限增长
- self_monitor.py Telegram告警无冷却，宕机时每30秒刷屏
- schedule_runner.py一次性调度执行超60秒会重复触发
- 限速脚本EXPIRE每次重置窗口可绕过（改用Lua原子脚本）

Security:
- JWT access token加token_version声明，改密码后旧token立失效（零宽限）
- INSTALL_MODE导入时常量→动态函数，安装后JWT认证不再残留禁用
- install.py /lock端点加管理员存在性验证，防止阻断安装
- ServerUpdate schema移除connectivity只读字段，防止伪造连接状态

Frontend fixes:
- doExec()缺r.ok检查、commands.html null检查
- _server_to_dict()补last_checked_at+ssh_key_public
- _field_match()逗号cron表达式修复
- alerts类型显示、SSH会话名称、搜索高亮定位
- 一次性/循环定时任务（run_mode+fire_at+自动禁用）

Dead code removed (400+ lines):
- SyncService batch_push/_push_single等5个方法（零调用者）
- 5个未使用schema（SyncCommands/SyncConfig/SyncSftp/FileDeploy/PaginatedResponse）
- 6个零调用service方法、3个无前端API端点
- 4个未使用import

Schema migrations:
- push_schedules: run_mode + fire_at列，cron_expr改NULL
- servers: 7个新列 + ssh_key_private/public VARCHAR(500)→TEXT

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

server/application/services/server_service.py

@@ -7,9 +7,8 @@ import json
 import logging
 from typing import Optional, List, Dict
 
-from server.domain.models import Server, SyncLog
+from server.domain.models import Server
 from server.domain.repositories import ServerRepository, SyncLogRepository
-from server.infrastructure.redis.client import get_redis
 
 logger = logging.getLogger("nexus.server_service")
 
@@ -21,12 +20,6 @@ class ServerService:
         self.server_repo = server_repo
         self.sync_log_repo = sync_log_repo
 
-    async def list_servers(self, category: Optional[str] = None) -> List[Server]:
-        """List all servers, optionally filtered by category"""
-        if category:
-            return await self.server_repo.get_by_category(category)
-        return await self.server_repo.get_all()
-
     async def get_server(self, id: int) -> Optional[Server]:
         return await self.server_repo.get_by_id(id)
 
@@ -47,25 +40,17 @@ class ServerService:
         await self.server_repo.update_heartbeat(id, is_online, system_info, agent_version)
 
     async def check_all_servers(self, server_ids: List[int]) -> Dict[int, dict]:
-        """Check health of multiple servers via Agent API (async concurrent)"""
-        import httpx
+        """Check health of multiple servers via SSH (no Agent port 8601 required)."""
+        from server.infrastructure.ssh.remote_probe import ssh_health_probe
+
         results = {}
 
         async def _check_one(server: Server) -> tuple:
-            from server.infrastructure.agent_url import agent_url
-            url = agent_url(server.domain, server.agent_port, "/health")
             try:
-                async with httpx.AsyncClient(timeout=5.0) as client:
-                    headers = {}
-                    if server.agent_api_key:
-                        headers["X-API-Key"] = server.agent_api_key
-                    resp = await client.get(url, headers=headers)
-                    if resp.status_code == 200:
-                        data = resp.json()
-                        return server.id, {"status": "online", "system_info": data}
-                    return server.id, {"status": "offline", "error": f"HTTP {resp.status_code}"}
+                probe = await ssh_health_probe(server, timeout=15)
+                return server.id, probe
             except Exception as e:
-                return server.id, {"status": "offline", "error": str(e)[:200]}
+                return server.id, {"status": "offline", "error": str(e)[:200], "channel": "ssh"}
 
         servers = []
         for sid in server_ids:
@@ -80,41 +65,4 @@ class ServerService:
             for sid, result in check_results:
                 results[sid] = result
 
-        return results
-
-    async def push_to_servers(
-        self,
-        server_ids: List[int],
-        source_path: str,
-        sync_mode: str = "incremental",
-        operator: str = "admin",
-        batch_size: int = 50,
-        concurrency: int = 10,
-    ) -> Dict[int, SyncLog]:
-        """Push files to multiple servers — delegates to SyncService.batch_push.
-
-        Note: API routes should use SyncService directly via Depends(get_sync_service).
-        This method exists for internal callers (e.g. schedule_runner) that already
-        have a ServerService instance.
-        """
-        from server.application.services.sync_service import SyncService
-
-        # Build a SyncService sharing our existing repos + the extra ones it needs
-        # NOTE: audit_repo and retry_repo are None here — this method is a compatibility
-        # shim for internal callers only. Do NOT call sync_service.batch_push() methods
-        # that invoke self.audit_repo or self.retry_repo, or they will raise AttributeError.
-        # API routes should use SyncService directly via Depends(get_sync_service).
-        sync_service = SyncService(
-            server_repo=self.server_repo,
-            sync_log_repo=self.sync_log_repo,
-            audit_repo=None,
-            retry_repo=None,
-        )
-        return await sync_service.batch_push(
-            server_ids=server_ids,
-            source_path=source_path,
-            sync_mode=sync_mode,
-            operator=operator,
-            batch_size=batch_size,
-            concurrency=concurrency,
-        )
+        return results