fix: 看门狗拉起后增加宽限期防重启循环

主程序重启后 JVM 启动 + DB 建连可能耗时数分钟，300 秒宽限期内探测失败只记录不累计，避免重启循环。

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

agent/dispatcher.py

@@ -9,7 +9,7 @@ B 端通过 WebSocket 下发 JSON 命令：{"type":"CMD_TYPE","payload":{...}}
 - GET_VERSION     查询主程序版本
 - GET_LOG_FILES   日志文件列表
 - LOG_TAIL        日志尾部（payload: file, lines）
-- LOG_TAIL_CTRL   日志实时流开关（payload: action, file）
+- GET_LOG_DOWNLOAD 日志全量下载（payload: file）
 - RESTART         重启主程序（优雅停机后由 ProcessGuard 拉起）
 - GET_CONFIG      读取医院配置文件（payload: hospital）
 - SAVE_CONFIG     写回医院配置文件（payload: hospital, content）
@@ -36,8 +36,6 @@ class CommandDispatcher:
 
         self._client = None                  # 由 main 注入（避免构造环）
         self._upgrade_command = None         # 升级命令（懒加载）
-        self._log_stream = None              # 日志实时流管理（懒加载）
-        self._log_stream_lock = threading.Lock()
         self._upgrade_lock = threading.Lock()
 
     def set_client(self, client):
@@ -52,18 +50,6 @@ class CommandDispatcher:
                 self._upgrade_command = UpgradeCommand(self.config, self.main_app_client, self._client)
             return self._upgrade_command
 
-    def _get_log_stream(self):
-        """懒加载日志流管理器。"""
-        with self._log_stream_lock:
-            if self._log_stream is None:
-                from agent.log_stream import LogStreamManager
-                self._log_stream = LogStreamManager(self.config, self.main_app_client, self._client)
-            return self._log_stream
-
-    def shutdown_log_stream(self):
-        """日志实时流（供 main 停机时统一释放）。"""
-        self._get_log_stream().stop_all()
-
     # ================= 分发 =================
 
     def dispatch(self, msg_type, payload):
@@ -89,11 +75,6 @@ class CommandDispatcher:
             lines = (payload or {}).get("lines") or 200
             result = {"ok": True, "data": self.main_app_client.log_tail(file, lines)}
 
-        elif msg_type == "LOG_TAIL_CTRL":
-            action = (payload or {}).get("action") or "start"
-            file = (payload or {}).get("file") or "info.log"
-            result = self._get_log_stream().control(action, file)
-
         elif msg_type == "RESTART":
             # 优先走 NSSM 服务重启（主程序假死时 HTTP 不可达也能重启成功）；
             # 未配置服务名时回退环回 HTTP 触发主程序优雅重启

agent/log_stream.py

@@ -1,141 +0,0 @@
-# -*- coding: utf-8 -*-
-"""日志实时推送管理器。
-
-B 端通过 LOG_TAIL_CTRL 命令开启/停止某设备某日志文件的实时流：
-- 开启后每 2 秒从主程序 /api/admin/log/tail 拉取增量行
-- 新行打包为 LOG_BATCH 上报 B 端（B 端转发给浏览器展示）
-- 每个文件独立游标（已发送行数），支持多文件并行
-
-增量算法：tail(max(offset+FETCH, 400)) 行，取 offset 之后的部分；
-日志滚动（文件变小/重置）时自动重置游标。
-"""
-
-import json
-import logging
-import threading
-import time
-
-log = logging.getLogger(__name__)
-
-# 轮询间隔（秒）
-POLL_SEC = 2
-# 单次拉取上限（行）
-FETCH_MAX = 400
-
-
-class LogStreamManager:
-    """日志实时流：单线程轮询全部活跃文件，增量上报 LOG_BATCH。"""
-
-    def __init__(self, config, main_app_client, client):
-        self.config = config
-        self.main_app_client = main_app_client
-        self.client = client
-
-        self._cursors: dict[str, int] = {}   # 文件名 → 已发送行数
-        self._active: dict[str, bool] = {}   # 文件名 → 是否激活
-        self._lock = threading.Lock()
-        self._stop = threading.Event()
-        self._thread: threading.Thread | None = None
-
-    # ================= 控制入口 =================
-
-    def control(self, action: str, file: str) -> dict:
-        """
-        控制命令入口。
-
-        :param action: start / stop
-        :param file:   日志文件名（info.log / error.log）
-        """
-        if action == "start":
-            with self._lock:
-                if self._active.get(file):
-                    return {"ok": True, "msg": f"流已存在：{file}"}
-                self._active[file] = True
-                self._cursors.pop(file, None)
-            self._ensure_thread()
-            log.info("日志流开启：%s", file)
-            return {"ok": True, "msg": f"日志流已开启：{file}"}
-
-        if action == "stop":
-            with self._lock:
-                self._active.pop(file, None)
-                self._cursors.pop(file, None)
-            log.info("日志流停止：%s", file)
-            return {"ok": True, "msg": f"日志流已停止：{file}"}
-
-        return {"ok": False, "msg": f"未知 action：{action}（支持 start/stop）"}
-
-    def stop_all(self):
-        """停止全部流（Agent 停机时调用）。"""
-        self._stop.set()
-        with self._lock:
-            self._active.clear()
-            self._cursors.clear()
-
-    def _ensure_thread(self):
-        """确保轮询线程在运行（首个流开启时启动）。"""
-        if self._thread is not None and self._thread.is_alive():
-            return
-        self._thread = threading.Thread(target=self._poll_loop, name="log-stream", daemon=True)
-        self._thread.start()
-
-    # ================= 轮询 =================
-
-    def _poll_loop(self):
-        """轮询主循环：每 POLL_SEC 秒拉一轮全部活跃文件。"""
-        while not self._stop.wait(POLL_SEC):
-            with self._lock:
-                files = list(self._active.keys())
-            for file in files:
-                if self._stop.is_set():
-                    return
-                try:
-                    self._poll_one(file)
-                except Exception as e:
-                    log.warning("日志流轮询失败（%s）：%s", file, e)
-
-    def _poll_one(self, file: str):
-        """单轮增量拉取：取尾部行，比对游标，新行打包上报。"""
-        with self._lock:
-            if not self._active.get(file):
-                return
-            offset = self._cursors.get(file, 0)
-
-        # 多拉一段，保证能覆盖到 offset 之前的已有内容
-        fetch = max(offset + 100, FETCH_MAX)
-        raw = self.main_app_client.log_tail(file, fetch)
-        log.debug("日志流轮询：file=%s, offset=%s, fetch=%s, rawLen=%s",
-                  file, offset, fetch, len(raw) if raw else -1)
-        if raw is None:
-            return
-        try:
-            node = json.loads(raw)
-        except Exception:
-            return
-        if node.get("code") != 0:
-            return
-        content = node.get("data")
-        if not content:
-            return
-
-        all_lines = content.split("\n")
-
-        # 日志滚动检测：行数比游标还少很多 → 文件已重置
-        if len(all_lines) < offset:
-            with self._lock:
-                self._cursors[file] = 0
-            return
-        if len(all_lines) <= offset:
-            return  # 无新行
-
-        fresh = [line for line in all_lines[offset:] if line and line.strip()]
-        with self._lock:
-            self._cursors[file] = len(all_lines)
-        if not fresh:
-            return
-
-        self.client.send("LOG_BATCH", {
-            "file": file,
-            "lines": fresh,
-            "ts": int(time.time()),
-        })

agent/main.py

@@ -89,10 +89,8 @@ def _setup_logging(log_dir: str):
 
 
 def _shutdown(signum, frame):
-    """优雅停机：释放日志流 → 停守护 → 断连接 → 退出。"""
+    """优雅停机：停守护 → 断连接 → 退出。"""
     log.info("Agent 停机中...（signal=%s）", signum)
-    if _dispatcher is not None:
-        _dispatcher.shutdown_log_stream()
     if _guard is not None:
         _guard.stop()
     if _client is not None:

agent/process_guard.py

@@ -9,6 +9,7 @@
 import logging
 import subprocess
 import threading
+import time
 
 log = logging.getLogger(__name__)
 
@@ -17,6 +18,11 @@ log = logging.getLogger(__name__)
 # 避免升级替换 jar 时被看门狗拉起的新进程抢占文件锁。
 _upgrading = threading.Event()
 
+# 拉起后的宽限期（秒）：主程序重启后 JVM 启动 + DB 建连可能耗时数分钟
+# （如机器休眠唤醒后网络未就绪，实测 4 分钟）。宽限期内探测失败只记录不累计，
+# 避免重启循环（重启→DB 慢→又判宕机→再重启）。
+GRACE_PERIOD_SEC = 300
+
 
 class ProcessGuard:
     """看门狗：周期健康探测 + 宕机拉起 + 告警上报。"""
@@ -28,6 +34,7 @@ class ProcessGuard:
 
         self._fail_count = 0       # 连续健康探测失败次数
         self._restart_count = 0    # 连续拉起次数
+        self._grace_until = 0.0    # 拉起后的宽限期截止时刻（monotonic 秒）
         self._stop = threading.Event()
         self._thread: threading.Thread | None = None
 
@@ -57,10 +64,19 @@ class ProcessGuard:
                 log.warning("守护探测异常：%s", e)
 
     def _check(self):
-        """单次探测：健康则清零计数；连续 3 次失败则拉起（升级期间跳过）。"""
+        """单次探测：健康则清零计数；连续 3 次失败则拉起（升级期间/宽限期内跳过）。"""
         health = self.main_app_client.health()
         if health.get("ok") is True:
             self._fail_count = 0
+            if self._grace_until and time.monotonic() < self._grace_until:
+                log.info("宽限期内主程序恢复健康，结束宽限")
+            self._grace_until = 0.0
+            return
+
+        # 拉起后宽限期内：主程序可能在启动/DB 建连中，只记录不累计失败
+        if self._grace_until and time.monotonic() < self._grace_until:
+            log.info("主程序拉起后宽限期内（剩 %d 秒），本次探测失败不累计：%s",
+                     int(self._grace_until - time.monotonic()), health.get("error"))
             return
 
         # 升级流程主动停的主程序，看门狗不接管，避免抢占 jar 锁
@@ -105,6 +121,8 @@ class ProcessGuard:
             else:
                 proc = start_main_app(self.config.main_start_cmd)
                 log.info("启动命令已执行，pid=%s", proc.pid if proc else None)
+            # 拉起后进入宽限期：JVM 启动 + DB 建连期间不累计探测失败，防重启循环
+            self._grace_until = time.monotonic() + GRACE_PERIOD_SEC
         except Exception as e:
             log.error("拉起失败：%s", e)
 

agent/upgrade.py

@@ -21,6 +21,7 @@ import shutil
 import subprocess
 import threading
 import time
+import zipfile
 
 import requests
 
@@ -118,6 +119,18 @@ class UpgradeCommand:
                 return self._fail(version, f"SHA-256 校验失败：expected={sha256}, actual={actual}")
             log.info("SHA-256 校验通过")
 
+        # ===== 2.5 解压 zip 得到内层 jar（上传端为绕过中间设备拦截，jar 被打成 zip） =====
+        extracted_jar = os.path.join("tmp", f"upgrade-{version}.jar.extracted")
+        try:
+            result = extract_jar_from_zip(tmp_jar, extracted_jar)
+            if result is None:
+                os.remove(tmp_jar)
+                return self._fail(version, "zip 包内未找到 jar 文件")
+            os.remove(tmp_jar)                    # 删除临时 zip
+            tmp_jar = extracted_jar               # 后续替换使用解压出的真 jar
+        except Exception as e:
+            return self._fail(version, f"解压 zip 失败：{e}")
+
         # ===== 3. 备份 =====
         # jar 路径来自配置（main.jar.path），适配不同部署结构
         #（如 Agent 与主程序分目录、jar 文件名带版本号等）
@@ -333,6 +346,19 @@ def copy_with_retry(src: str, dst: str, retries: int):
     ) from last_error
 
 
+def extract_jar_from_zip(zip_path: str, dst_jar: str) -> str | None:
+    """从 zip 中取出第一个 .jar 文件写到 dst_jar，返回 dst_jar；找不到返回 None。"""
+    with zipfile.ZipFile(zip_path) as z:
+        jars = [n for n in z.namelist()
+                if n.lower().endswith(".jar") and not n.endswith("/")]
+        if not jars:
+            return None
+        with z.open(jars[0]) as src, open(dst_jar, "wb") as dst:
+            shutil.copyfileobj(src, dst)
+        log.info("解压 zip 得到 jar：%s → %s", jars[0], os.path.basename(dst_jar))
+        return dst_jar
+
+
 def sha256_file(path: str) -> str:
     """文件 SHA-256（分块读取，支持大文件）。"""
     digest = hashlib.sha256()