feat(run): PID 存活检测 — 死掉的 run 显示「已中断」而非永远运行中

根因:run 被硬杀(SIGKILL/OOM/重启)时来不及写终态,DB status 永远停在 running。
方案(无时间阈值):开跑时把执行进程身份(pid+host+/proc starttime)写进 lifecycle_start 事件;
读状态时对 running 的 run 查那个进程是否还活着(os.kill(pid,0)+starttime 防 PID 复用),
死了 → 显示 interrupted。取不到/跨机/旧 run → 保持原状,绝不误杀真实运行中的 run。
前端 RunListPage 加「已中断」(红)。仅对新 run 生效(旧 run 无 executor 记录)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

content_agent/dashboard_service.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import os
+import socket
 from datetime import date, datetime
 from pathlib import Path
 from typing import Any
@@ -76,7 +78,7 @@ class DashboardService:
                 page_size=page_size,
             )
         items = [
-            self._local_run_list_item(run_id)
+            self._apply_liveness(self._local_run_list_item(run_id))
             for run_id in getattr(self.runtime, "list_runs", lambda: [])()
         ]
         items = [
@@ -382,7 +384,7 @@ class DashboardService:
             tuple([*params, page_size, (page - 1) * page_size]),
         )
         return {
-            "items": [_json_safe(_db_run_row_to_item(row)) for row in rows],
+            "items": [self._apply_liveness(_json_safe(_db_run_row_to_item(row))) for row in rows],
             "page": page,
             "page_size": page_size,
             "total": int((count_row or {}).get("cnt") or 0),
@@ -399,7 +401,29 @@ class DashboardService:
             "FROM `content_agent_runs` r WHERE r.run_id = %s LIMIT 1",
             (run_id,),
         )
-        return _json_safe(_db_run_row_to_item(row or {"run_id": run_id, "status": "unknown"}))
+        return self._apply_liveness(_json_safe(_db_run_row_to_item(row or {"run_id": run_id, "status": "unknown"})))
+
+    def _apply_liveness(self, item: dict[str, Any]) -> dict[str, Any]:
+        # status=running 但执行进程已死(硬杀/OOM/重启,来不及写终态)→ 显示「已中断」。无时间阈值。
+        if item.get("status") != "running":
+            return item
+        events = self._read_jsonl_optional(item.get("run_id") or "", "run_events.jsonl") or []
+        executor = next(
+            (
+                row["raw_payload"]["executor"]
+                for row in events
+                if isinstance(row.get("raw_payload"), dict)
+                and isinstance(row["raw_payload"].get("executor"), dict)
+            ),
+            None,
+        )
+        if not executor:
+            return item  # 旧 run 没记 executor,无法判定,保持原状
+        if executor.get("host") and executor.get("host") != socket.gethostname():
+            return item  # 跨机,本地查不了进程
+        if not _process_alive(executor.get("pid"), executor.get("proc_start")):
+            return {**item, "status": "interrupted", "current_step": item.get("current_step") or "unknown"}
+        return item
 
     def _local_run_list_item(self, run_id: str) -> dict[str, Any]:
         final_output = self._read_json_optional(run_id, "final_output.json") or {}
@@ -553,6 +577,36 @@ def _db_run_filters(filters: dict[str, Any]) -> tuple[str, list[Any]]:
     return "WHERE " + " AND ".join(clauses), params
 
 
+def _proc_start_time(pid: int) -> str | None:
+    try:
+        with open(f"/proc/{pid}/stat", encoding="utf-8") as handle:
+            content = handle.read()
+        return content[content.rindex(")") + 2:].split()[19]
+    except Exception:
+        return None
+
+
+def _process_alive(pid: Any, expected_start: Any) -> bool:
+    # True = 活着 / 无法判定(绝不误杀真实运行中的 run);False = 进程确实没了(或 PID 被复用)。
+    try:
+        pid_int = int(pid)
+    except (TypeError, ValueError):
+        return True
+    try:
+        os.kill(pid_int, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True  # 进程在,只是不归当前用户
+    except Exception:
+        return True
+    if expected_start:
+        actual = _proc_start_time(pid_int)
+        if actual is not None and actual != str(expected_start):
+            return False  # PID 被复用成了别的进程
+    return True
+
+
 def _db_run_row_to_item(row: dict[str, Any]) -> dict[str, Any]:
     return {
         "run_id": row.get("run_id"),

content_agent/run_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import socket
 import threading
 from collections import Counter
 from datetime import datetime, timezone
@@ -139,7 +140,7 @@ class RunService:
                 event_type="run_started",
                 status="running",
                 message="run started",
-                raw_payload={"source_ref": resolved_source_ref},
+                raw_payload={"source_ref": resolved_source_ref, "executor": _executor_identity()},
             )
 
             # 先验证/构造平台 client(拒绝非法平台),再过 Gate 1(不为非法平台浪费 PG 调用)。
@@ -779,6 +780,23 @@ def _utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+def _proc_start_time(pid: int) -> str | None:
+    # Linux /proc/<pid>/stat 第 22 字段 starttime(自启动以来的时钟滴答),作为 PID 复用的指纹;
+    # comm(第 2 字段)在括号里且可能含空格,故按最后一个 ')' 切分。非 Linux/取不到 → None。
+    try:
+        with open(f"/proc/{pid}/stat", encoding="utf-8") as handle:
+            content = handle.read()
+        return content[content.rindex(")") + 2:].split()[19]
+    except Exception:
+        return None
+
+
+def _executor_identity() -> dict[str, Any]:
+    # 记录"谁在跑这条 run":pid + 主机 + 进程启动时刻。读状态时据此判进程是否还活着(无时间阈值)。
+    pid = os.getpid()
+    return {"pid": pid, "host": socket.gethostname(), "proc_start": _proc_start_time(pid)}
+
+
 def _load_project_env(env_file: Path | str = ".env") -> dict[str, str]:
     path = Path(env_file)
     if not path.exists():

tests/test_run_liveness.py

@@ -0,0 +1,27 @@
+"""PID 存活检测:死进程的 running run 应判为 interrupted,绝不误杀真实运行中的 run。"""
+
+import os
+
+from content_agent.dashboard_service import _process_alive
+
+
+def test_current_process_is_alive():
+    assert _process_alive(os.getpid(), None) is True
+
+
+def test_nonexistent_pid_is_dead():
+    assert _process_alive(999999, None) is False
+
+
+def test_unknown_pid_not_flagged():
+    # 取不到/解析不了 pid 时,宁可显示运行中也不误判。
+    assert _process_alive(None, None) is True
+    assert _process_alive("not-an-int", None) is True
+
+
+def test_pid_reuse_detected_via_start_time():
+    # 仅 Linux 有 /proc:活进程但启动时刻对不上 → 视为 PID 被复用 → 判死。
+    if not os.path.exists(f"/proc/{os.getpid()}/stat"):
+        return
+    assert _process_alive(os.getpid(), "0") is False
+    assert _process_alive(os.getpid(), None) is True

web2/features/RunListPage.tsx

@@ -13,6 +13,8 @@ function statusLabel(value?: string | null): string {
     completed: "已完成",
     running: "运行中",
     failed: "运行失败",
+    interrupted: "已中断",
+    blocked: "已拦截",
     pending: "等待处理",
     success: "已完成",
     partial_success: "已完成"
@@ -20,6 +22,13 @@ function statusLabel(value?: string | null): string {
   return value ? labels[value] || "状态待确认" : "状态待确认";
 }
 
+function statusChipClass(value?: string | null): string {
+  if (value === "failed" || value === "interrupted") return "red";
+  if (value === "running" || value === "pending") return "blue";
+  if (value === "blocked") return "yellow";
+  return "green";
+}
+
 function formatBeijingTime(value?: string | null): string {
   if (!value) return "未记录开始时间";
   const iso = /z$|[+-]\d{2}:\d{2}$/i.test(value) ? value : `${value}Z`;
@@ -75,7 +84,7 @@ export function RunListPage() {
             </div>
             <div className="run-row-meta">
               <span className="chip">{platformLabel(run.platform)}</span>
-              <span className="chip blue">{statusLabel(run.status)}</span>
+              <span className={`chip ${statusChipClass(run.status)}`}>{statusLabel(run.status)}</span>
               <span>{formatBeijingTime(run.started_at)}</span>
             </div>
           </Link>