整点在线推 Telegram，并带上搬运管道的新鲜度

执行器活着不代表链路活着——搬运死了一样心跳正常、一样什么都不做。
所以每小时那条必须读 ship_alive.json：ssh 是否在连、文件是否还在刷。
连上/断开立刻落盘，不靠 5 分钟心跳，否则第一轮会误报上游断了。

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

live/ship_signals.py

@@ -72,6 +72,8 @@ class Shipper:
         self.last_signal_ts = 0.0
         self.n_reconnect = 0
         self.n_skew = 0
+        self.ssh_up = False
+        self.alive = local_bus.parent / "ship_alive.json"
 
     def load_seen(self) -> None:
         """本地已有的键先读进来，避免重启后把整个文件再追加一遍。"""
@@ -127,6 +129,9 @@ class Shipper:
             *cmd, stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE)
         self.connected_at = time.time()
+        self.ssh_up = True
+        self.touch_alive(0)          # 立刻落盘，别等 5 分钟心跳——执行器第一轮
+                                     # 整点推送会读这个文件，晚写就会误报上游断了
         print(f"  ssh 已连上 {self.host}", flush=True)
         assert proc.stdout is not None
         try:
@@ -143,6 +148,9 @@ class Shipper:
                 proc.kill()
             await proc.wait()
             up = time.time() - self.connected_at
+            self.ssh_up = False
+            self.touch_alive(up)     # 立刻标断开。只靠停更来发现的话，心跳还在
+                                    # 刷 ts，执行器会以为管道还活着
             msg = err.decode("utf-8", "replace").strip()
             print(f"  ssh 断开（在线 {up:.0f}s，退出码 {proc.returncode}）"
                   f"{'：' + msg if msg else ''}", flush=True)
@@ -177,6 +185,30 @@ class Shipper:
                   f"{self.n_reconnect} 次 · 新增 {self.n_new} 条"
                   f"（重放去重 {self.n_dup}）· 最近一条 {last}{skew}",
                   flush=True)
+            self.touch_alive(up)
+
+    def touch_alive(self, up: float) -> None:
+        """把连接状态落到文件，供执行器的整点推送读。
+
+        为什么要落盘：Telegram 推送在执行器那侧，而它看不到本进程的日志。
+        「执行器活着」单独没有意义——搬运管道死掉时执行器一样心跳正常、一样
+        什么都不做，那正是最危险的状态。所以推送里必须带上游的新鲜度，
+        这个文件是唯一的传递途径。
+
+        写失败只打日志：搬运的正事是投信号，不能因为写不了状态文件而中断。
+        """
+        try:
+            self.alive.parent.mkdir(parents=True, exist_ok=True)
+            tmp = self.alive.with_suffix(".tmp")
+            tmp.write_text(json.dumps({
+                "ts": time.time(), "up_s": round(up),
+                "connected": self.ssh_up,
+                "n_reconnect": self.n_reconnect, "n_new": self.n_new,
+                "n_skew": self.n_skew,
+                "last_signal_ts": self.last_signal_ts}), encoding="utf-8")
+            tmp.replace(self.alive)          # 原子替换，读侧不会看到半个文件
+        except Exception as e:               # noqa: BLE001
+            print(f"  ⚠ 写存活文件失败 {type(e).__name__}: {e}", flush=True)
 
 
 def main() -> None: