dsh_ai1net_server/归档/技能包快照/session-mechanism-20261004/scripts/session-rules-check.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""session-rules-check.py —— 「会话规则机制体检」＝ 开会话后的第一件事

用户口径（2026-10-02 原话逐字，含一次纠正）：
  ① 「这个会话和协作会话的技能包 运行的第一件事 ，就应该是检查清楚
      所有会话规划是否配置完整且生效，然后标记一个状态」
  ② 「就应该是检查清楚 **所有会话规则机制** 是否配置完整且生效，**不是规划 是 规则**」
  ⇒ 对象是「**规则机制**」（钩子 / 闸门 / 技能指针 / 常驻 / 编排…），⛔ 不是"排期规划"。

为什么要它（真因，⛔ 不是想当然）
──────────────────────────────
2026-10-02 一天里连续查出**四例「配置在册、其实没生效」**：
  · 钩子注入文本指着 **09-28 就已合并退役**的技能名（命中后让人去加载不存在的东西）；
  · 常驻规则快照的生成脚本**写到没人读的幽灵目录** ⇒ 快照停在 09-28、10-01 的定稿进不去；
  · 工作区记忆（**每轮注入**的那份）里 3 处指针**指向已退役技能名**；
  · 锚点校验拿**旧版逐字短语**当探针 ⇒ 每次报 4 处「规则丢了」的**假缺失**。
⇒ 共同点：**"在册" ≠ "生效"**。这类状态**不问就不会知道**，等它表现成"卡住"就晚了。
⇒ 所以开工第一件事＝**把规则的"配置"与"生效"两件事都问清楚，并落一枚状态标记**。

检查三类、十二项（⛔ 每项都必须**能真报出问题** —— 见文末变异对照记录）：
  A 机制装没装好    ① 关键钩子在册 ② 钩子脚本路径存在
                     ③ **钩子注入里引用的技能名是否都存在** ④ 钩子**真在被调用**没（闸门日志新鲜度）
  B 规则载体同没同步 ⑤ **每轮注入的记忆里引用的技能名是否都存在** ⑥ 常驻规则快照是否**比权威文件旧**
  C 编排在不在跑    ⑦ 唤醒/跟进 两台周期钟 ⑧ 排期模型可用性（`thinking=0`+flash ⇒ 必被拒）
                     ⑨ `cwds` **归属同形**（错一字面 ⇒ 裂组且自我强化） ⑩ 投递心跳
                     ⑪ 三类会话有活的没 ⑫ 有没有「`once` 且**从未运行过**」就失效的排期

产出：`<WS>/.workbuddy/collab/session-rules.json`（＝用户要的那枚「状态标记」）＋ stdout 摘要
      ⚠️ stdout **只打非 ok 项**（本脚本每轮都会随 `state.py` 跑 ⇒ 输出本身就是成本）
退出码：0 = 全 ok ｜ 1 = 有 warn/fail（⛔ **只标记，不拦开工**）
用法：python session-rules-check.py [--ws <工作区>] [--json] [--all]
"""
import ast
import io
import json
import os
import re
import sqlite3
import sys
import time

# ── 根目录外置（roots.env 由 install.py 生成；宿主 env 优先，本文件兜底） ──────
def _sm_load_roots():
    here = os.path.dirname(os.path.abspath(__file__))
    for up in range(4):
        p = os.path.normpath(os.path.join(here, *([".."] * up), "roots.env"))
        if os.path.isfile(p):
            try:
                for ln in io.open(p, encoding="utf-8"):
                    ln = ln.strip()
                    if ln and not ln.startswith("#") and "=" in ln:
                        k, v = ln.split("=", 1)
                        os.environ.setdefault(k.strip(), v.strip())
            except Exception:
                pass
            return
_sm_load_roots()

ARGV = sys.argv[1:]
def _opt(name, default=None):
    if name in ARGV:
        i = ARGV.index(name)
        return ARGV[i + 1] if i + 1 < len(ARGV) else default
    return default

WS = (_opt("--ws") or os.environ.get("DSH_COLLAB_WS") or os.environ.get("DSH_WS_ROOT")
      or os.getcwd()).replace("\\", "/").rstrip("/")
CFG = os.environ.get("CODEBUDDY_CONFIG_DIR") or os.path.join(os.path.expanduser("~"), ".workbuddy")
DB = os.path.join(CFG, "workbuddy.db")
SETTINGS = os.path.join(CFG, "settings.json")
SKILLS = os.environ.get("DSH_SKILLS_ROOT") or os.path.join(CFG, "skills")
OUT = WS + "/.workbuddy/collab/session-rules.json"
AS_JSON = "--json" in ARGV
SHOW_ALL = "--all" in ARGV

# 「会话机制三件套」的**关键钩子**（缺一 ⇒ 该机制不会运行）
KEY_HOOKS = [
    ("SessionStart", "lock-guard-hook.py", "无锁不许改代码库 / 文档库"),
    ("PreToolUse", "lock-guard-hook.py", "同上（Write|Edit 面）"),
    ("PreToolUse", "bash-output-guard.py", "拦下会灌爆上下文的读命令"),
    ("UserPromptSubmit", "wb-result-hook.py", "收结果 / 结果回流"),
    ("UserPromptSubmit", "stop-dialog-guard.py", "水位与收口（接续机制起点）"),
    ("UserPromptSubmit", "skill-load-guard.py", "用户点名方法 ⇒ 强制加载技能"),
    ("UserPromptSubmit", "session-log-guard.py", "会话日志闸（防把界面顶死）"),
    ("PostToolUse", "session-log-guard.py", "同上（工具后）"),
]
GATE_LOGS = [("收口", WS + "/.workbuddy/stop-dialog-guard.log"),
             ("技能", WS + "/.workbuddy/skill-load-guard.log"),
             ("限流", WS + "/.workbuddy/bash-guard.log"),
             # ⚠️ 锁日志落在**文档库的父目录**（lock-guard-hook 管的是文档库 / 代码库，不是本工作区）
             #    ⇒ 首版按 skills 目录往上推两层 ⇒ 算成 `E:/ProgramData/.workbuddy/…` ⇒ **假"缺失"**。
             ("锁", (os.path.join(os.path.dirname(os.environ.get("DSH_DOCS_ROOT", "")),
                                  ".workbuddy", "lock-hook.log")
                     if os.environ.get("DSH_DOCS_ROOT") else ""))]
ROLES_LIVE = [("唤醒", "[唤醒]"), ("协作", "[协作]"), ("跟进", "[跟进]")]
ROLES_CLOCK = [("唤醒", "[唤醒]"), ("跟进", "[跟进]")]   # ⚠️ 协作按需创建 ⇒ ⛔ 不要求它有周期钟
FLASH_HINT = ("flash",)
# 技能名形态（与「技能库引用体检」同源）；⚠️ 只认这几个前缀，别把业务线名当技能
SKILL_NAME_RE = re.compile(r"\b((?:dsh|agent|session|workbuddy|multi|third-party|stage|product|content-source)"
                           r"[a-z0-9]*(?:-[a-z0-9]+)+)\b")
# 「指向技能」的上下文（⛔ 只在这类上下文里判，否则会把业务线名 / localStorage 键当技能名）
#
# 🔴 2026-10-02 收紧（首版太宽、当场三条误报，全是"把不属于技能的东西当技能名"）：
#   旧版还含 `^[`「]?\s*$`（"整个字符串就是一个技能名形态"）⇒ 于是：
#     · `lock-guard-hook.py` 里的字符串 `"dsh-server-docs"`（那是**文档库目录名**）被报成悬空技能；
#     · `session-log-guard.py` 里的 `"session-log-guard"`（那是**它自己的脚本名**）同样被报；
#     · `wb-result-hook.py` 里的 `os.path.join(_base,"skills","multi-session-collab",…)`
#       （那是**故意留的旧位置回退候选**，代码注释已标明）也被报。
#   ⇒ 本项要问的其实只有一件事：**钩子在叫 AI「去加载」某个技能时，那个技能还在不在**
#     ⇒ 判据就只认「加载 / Skill 工具」这一类**动作语境**，⛔ 不认"字符串长得像技能名"。
SKILL_CTX_RE = re.compile(r"(?:Skill\s*工具(?:加载|来加载)|调用\s*Skill|先加载|加载|load)\s*[`「]?\s*$")

CHECKS = []
def rec(cid, group, level, title, detail=""):
    CHECKS.append({"id": cid, "group": group, "level": level, "title": title, "detail": detail})

# ── 工具 ───────────────────────────────────────────────────────────────
def _ro_conn():
    c = sqlite3.connect("file:%s?mode=ro" % DB.replace("\\", "/"), uri=True, timeout=8)
    c.execute("PRAGMA busy_timeout=6000")
    return c

def _norm_cwd(p):
    """与宿主去重键同源：`path.trim().toLowerCase()`（⛔ 别自作聪明归一成 basename）。"""
    return str(p or "").replace("\\", "/").strip().lower()

def _safe_list(v):
    if v is None:
        return []
    if isinstance(v, (list, tuple)):
        return [str(x) for x in v]
    s = str(v).strip()
    if s.startswith("["):
        try:
            r = json.loads(s)
            return [str(x) for x in r] if isinstance(r, list) else [s]
        except Exception:
            return [s]
    return [x.strip() for x in s.split(",") if x.strip()]

def _string_literals(src):
    """只取**运行时会用到的字符串字面量** ⇒ 排除注释与 docstring。

    为什么要这么讲究：钩子脚本头部**注释里大量出现旧技能名**（讲历史/讲事故），
    那是**合规保留**的（改了就是篡改取证记录）。若连注释一起扫 ⇒ 每次都报一堆假阳性 ⇒ 真问题被淹掉。
    """
    out = []
    try:
        tree = ast.parse(src)
    except Exception:
        return out
    doc_ids = set()
    for node in ast.walk(tree):
        if isinstance(node, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            body = getattr(node, "body", [])
            if (body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant)
                    and isinstance(body[0].value.value, str)):
                doc_ids.add(id(body[0].value))
    for node in ast.walk(tree):
        if isinstance(node, ast.Constant) and isinstance(node.value, str) and id(node) not in doc_ids:
            out.append(node.value)
    return out

def _dangling_skills_in_text(txts):
    """在一组字符串里找「指向不存在的技能名」。返回 {name: 片段}。"""
    out = {}
    for t in txts:
        for m in SKILL_NAME_RE.finditer(t):
            nm = m.group(1)
            if nm in EXISTING_SKILLS:
                continue
            head = t[max(0, m.start() - 14):m.start()]
            if not SKILL_CTX_RE.search(head):
                continue
            out.setdefault(nm, t[max(0, m.start() - 20):m.end() + 20].replace("\n", " ").strip())
    return out

EXISTING_SKILLS = set()
try:
    EXISTING_SKILLS = {n for n in os.listdir(SKILLS) if os.path.isdir(os.path.join(SKILLS, n))}
except Exception:
    pass


# ── A. 机制装没装好 ─────────────────────────────────────────────────────
def check_hooks():
    if not os.path.isfile(SETTINGS):
        rec("hook_reg", "A", "fail", "读不到 hooks 配置", "settings.json 不在：%s" % SETTINGS)
        return [], None
    try:
        hk = (json.load(io.open(SETTINGS, encoding="utf-8")) or {}).get("hooks") or {}
    except Exception as e:
        rec("hook_reg", "A", "fail", "hooks 配置解析失败", str(e)[:140])
        return [], None
    cmds = [(ev, str(h.get("command") or ""))
            for ev, arr in hk.items() for blk in (arr or []) for h in ((blk or {}).get("hooks") or [])]
    miss = [k for k in KEY_HOOKS if not any(k[0] == ev and k[1] in c for ev, c in cmds)]
    if miss:
        rec("hook_reg", "A", "fail", "关键钩子不在册（该机制现在不会运行）",
            "、".join("%s(%s)" % (k[1], k[0]) for k in miss))
    else:
        rec("hook_reg", "A", "ok", "关键钩子在册", "共 %d 条钩子注册" % len(cmds))

    bad_path = []
    for ev, c in cmds:
        m = re.search(r'([A-Za-z]:[\\/][^"\']*?\.(?:py|sh|mjs))', c)
        if m and not os.path.exists(m.group(1)):
            bad_path.append("%s(%s)" % (os.path.basename(m.group(1)), ev))
    if bad_path:
        rec("hook_path", "A", "fail", "钩子脚本路径不存在 ⇒ 静默失效", "、".join(sorted(set(bad_path))[:5]))
    else:
        rec("hook_path", "A", "ok", "钩子脚本路径均存在", "")
    return cmds, hk


def check_hook_skillnames(cmds):
    """🔴 本体检最有价值的一项：钩子**注入文本**里引用的技能名，现在还存在吗？

    真因（2026-10-02 实测）：`skill-load-guard.py` 注入「先调用 Skill 工具加载 `dsh-decision-method`」，
    而该技能 **2026-09-28 已合并退役**（现名 `dsh-decision`）⇒ 命中后让人去拿一份不存在的东西。
    """
    dangling = {}
    scanned = 0
    for ev, c in cmds:
        m = re.search(r'([A-Za-z]:[\\/][^"\']*?\.py)', c)
        if not m:
            continue
        p = m.group(1)
        if not os.path.isfile(p):
            continue
        try:
            src = io.open(p, encoding="utf-8", errors="replace").read()
        except Exception:
            continue
        scanned += 1
        for nm, frag in _dangling_skills_in_text(_string_literals(src)).items():
            dangling.setdefault("%s → %s" % (os.path.basename(p), nm), frag)
    if dangling:
        rec("hook_skillname", "A", "fail", "钩子注入里指向**已不存在的技能名**（命中即空转）",
            "；".join("%s" % k for k in sorted(dangling)[:4]))
    else:
        rec("hook_skillname", "A", "ok", "钩子注入引用的技能名均存在", "扫了 %d 份钩子脚本" % scanned)


def check_gate_logs():
    stale = []
    rows = []
    for tag, p in GATE_LOGS:
        try:
            with io.open(p, "rb") as f:
                f.seek(0, 2)
                n = min(8192, f.tell())
                f.seek(-n, 2)
                r = [x for x in f.read().decode("utf-8", "replace").split("\n") if x.strip()]
            if not r:
                stale.append(tag + "(空)")
                continue
            last = r[-1][:16]
            rows.append("%s=%s" % (tag, last))
            age = time.time() - os.path.getmtime(p)
            if age > 7 * 24 * 3600:
                stale.append("%s(停 %.0f 天)" % (tag, age / 86400.0))
        except Exception:
            stale.append(tag + "(缺)")
    if stale:
        rec("gate_fresh", "A", "warn", "部分闸门日志陈旧 / 缺失（可能已掉线）",
            "；".join(stale) + "｜" + "  ".join(rows))
    else:
        rec("gate_fresh", "A", "ok", "各闸门最近都被调用过", "  ".join(rows))


# ── B. 规则载体同没同步 ─────────────────────────────────────────────────
def check_memory_pointers():
    """工作区 `MEMORY.md` 是**每轮注入**的那份 ⇒ 它里面的悬空指针会一直把 AI 引向不存在的东西。"""
    p = os.path.join(WS, ".workbuddy", "memory", "MEMORY.md")
    if not os.path.isfile(p):
        rec("mem_ptr", "B", "warn", "读不到工作区 MEMORY.md", "路径：%s" % p)
        return
    try:
        txt = io.open(p, encoding="utf-8", errors="replace").read()
    except Exception as e:
        rec("mem_ptr", "B", "warn", "读 MEMORY.md 失败", str(e)[:120])
        return
    bad = {}
    for m in SKILL_NAME_RE.finditer(txt):
        nm = m.group(1)
        if nm in EXISTING_SKILLS:
            continue
        head = txt[max(0, m.start() - 14):m.start()]
        if not (SKILL_CTX_RE.search(head) or "技能" in head):
            continue
        bad.setdefault(nm, txt[max(0, m.start() - 26):m.end() + 12].replace("\n", " ").strip())
    if bad:
        rec("mem_ptr", "B", "fail", "工作区记忆里的技能指针悬空（每轮注入 ⇒ 一直把人引错）",
            "；".join(sorted(bad)[:4]))
    else:
        rec("mem_ptr", "B", "ok", "工作区记忆的技能指针均有效", "")


def check_snapshot_stale():
    """常驻规则快照 vs 权威 `CODEBUDDY.md`：**比权威旧 ⇒ 内容过期**（快照是副本，权威单向）。"""
    snap = os.path.join(SKILLS, "dsh-local-env", "references", "dsh-env-bootstrap",
                        "常驻规则-快照.md")
    auth = os.path.join(WS, "CODEBUDDY.md")
    if not os.path.isfile(auth):
        rec("snap_sync", "B", "warn", "找不到权威规则文件", auth)
        return
    if not os.path.isfile(snap):
        rec("snap_sync", "B", "warn", "常驻规则快照不存在", snap)
        return
    try:
        d = (os.path.getmtime(auth) - os.path.getmtime(snap)) / 86400.0
    except Exception:
        return
    if d > 0.01:
        rec("snap_sync", "B", "fail", "常驻规则快照**比权威文件旧** ⇒ 新规则没进快照",
            "权威 %s 比快照新 %.1f 天；重生成 ⇒ `python scripts/resident-rules.py --snapshot`"
            % (os.path.basename(auth), d))
    else:
        rec("snap_sync", "B", "ok", "常驻规则快照不比权威旧", "")


# ── C. 编排在不在跑 ─────────────────────────────────────────────────────
def _cwds_nearmiss(autos, ws_norm):
    """「与本工作区**几乎同名**、但字面不同」的排期 cwds ⇒ 会**裂成两组**。

    ⚠️ 判据只认**近失配**两条（⛔ 不是"同父目录就算" —— 首版就栽在这儿）：
      (a) **同父目录 ＋ 名字只是 `-`/`_`/大小写之别** —— 如 `…/ai1net_dsh_server`
          ↔ `…/ai1net-dsh-server`；⇒ **去标点后逐字相同** 才算近失配（例：`ai1net-decision-laya`
          与本工作区同父目录，但去标点后不同 ⇒ **是另一条线，⛔ 不是失配**）
      (b) **同名字、父目录不同** —— 如 `E:/…/ai1net-dsh-server` ↔ `D:/…/ai1net-dsh-server`
          （多半是从别的机器抄来的排期）
    ⇒ 去重键＝`path.trim().toLowerCase()`（宿主原样）⇒ 这两种各裂一组，且**自我强化**
      （越裂越不像，之后再也归不回来）⇒ 必须**在建的时候**就报出来。
    ⇒ 纯函数：只吃数据，便于用合成样本做红绿对照（⛔ 不靠"实跑一次看着对"）。
    """
    def _key(p):
        return re.sub(r"[^a-z0-9]", "", p)
    base = os.path.basename(ws_norm.rstrip("/"))
    parent = os.path.dirname(ws_norm.rstrip("/"))
    out = []
    for a in autos:
        for x in _safe_list(a.get("cwds")):
            s = str(x or "").strip()
            if not s:
                continue
            n = _norm_cwd(s).rstrip("/")
            if n == ws_norm:
                continue
            same_parent = (n.rsplit("/", 1)[0] if "/" in n else "") == parent
            name = n.rsplit("/", 1)[-1]
            if (same_parent and _key(name) == _key(base)) or (not same_parent and name == base):
                out.append("%s → %s" % ((a.get("name") or "")[:24], s))
    return out


def _sat_epoch(sat):
    """`scheduled_at` → epoch 秒。⚠️ 实测只到**分钟**（`2026-10-02T14:26`）⇒ 两种格式都试；
    解析不出来返回 `None`（⛔ 调用方据此走"不可核对"，**不猜**）。"""
    s = str(sat or "")
    for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M"):
        try:
            return time.mktime(time.strptime(s[:len(time.strftime(fmt))], fmt))
        except Exception:
            continue
    return None


def _once_zombie(mine, ran, grace_min=30.0):
    """`once` 排期、已无下次触发、却**从未运行过** ⇒ 那条活会**静默消失**。

    🔴 两条判据要点（都是实测校正出来的，⛔ 别想当然）：
      (a) **"跑完了"不是问题、"从未跑却已失效"才是** —— `once` 到点跑完即被消耗，属**正常痕迹**；
          把两者混报 ⇒ 首版拿 `last_run_at` 判，把 **18 条正常痕迹**当成 18 个故障。
      (b) **"有没有跑过"必须查 `automation_runs`，⛔ 不能查 `automations.last_run_at`** ——
          该字段宿主**根本不写**（实测：本会话自己那条排期**明明在跑**，值仍是 `None`）。
    🔴 2026-10-02 修（S11 · 与 `collabd.health()` 同一病灶）：本机 once 排期**不到点即时触发**
      而是主机轮询**补跑**（`runKind=missed`，实测延迟 0~11 分钟）⇒ **刚过点不足 `grace_min` 的
      那条不算「从未运行」**，只是**还在补跑窗口里**。⛔ 无 `scheduled_at` 或解析不出 ⇒ 不判 fail
      （不可核对 ≠ 有故障），归入第三返回值。
    ⇒ 纯函数：`ran` 传 `None` 表示"读不到运行记录" ⇒ **只报 warn，⛔ 不报 fail**。
    返回 (never_ran, consumed, unverifiable, in_grace)。
    """
    never, consumed, unver, in_grace = [], 0, 0, 0
    now = time.time()
    for a in mine:
        if a.get("schedule_type") != "once" or a.get("next_run_at"):
            continue
        if ran is None:
            unver += 1
            continue
        if a.get("id") in ran:
            consumed += 1
            continue
        # 🔴 到点后仍可能补跑 ⇒ 未超容差先归入"窗口内"，⛔ 不当故障
        t = _sat_epoch(a.get("scheduled_at"))
        if t is not None and (now - t) / 60.0 <= float(grace_min):
            in_grace += 1
            continue
        never.append((a.get("name") or "")[:30])
    return never, consumed, unver, in_grace


def check_orchestration():
    if not os.path.isfile(DB):
        rec("clock", "C", "fail", "读不到宿主库", DB)
        return
    try:
        c = _ro_conn()
        autos = [{"id": r[0], "name": r[1], "schedule_type": r[2], "next_run_at": r[3],
                  "model_id": r[4], "model_is_thinking": r[5], "cwds": r[6], "scheduled_at": r[7]}
                 for r in c.execute(
                     "select id,name,schedule_type,next_run_at,model_id,model_is_thinking,cwds,"
                     "       scheduled_at "
                     "from automations where deleted_at is null")]
        sess = list(c.execute("select id,title,status from sessions "
                              "where title like '[唤醒]%' or title like '[协作]%' or title like '[跟进]%'"))
    except Exception as e:
        rec("clock", "C", "fail", "读宿主库失败", str(e)[:140])
        return
    ws_norm = _norm_cwd(WS)
    mine = [a for a in autos if any(_norm_cwd(x) == ws_norm for x in _safe_list(a["cwds"]))]

    # ⑦ 周期钟
    clocks = {}
    for a in mine:
        if a["schedule_type"] != "recurring":
            continue
        for lb, pre in ROLES_CLOCK:
            if (a["name"] or "").startswith(pre):
                clocks.setdefault(lb, []).append(a)
    missing = [lb for lb, _ in ROLES_CLOCK if not clocks.get(lb)]
    if missing:
        rec("clock", "C", "fail", "周期钟缺失 ⇒ 没人在推 / 收", "、".join(missing))
    else:
        rec("clock", "C", "ok", "周期钟在册",
            "；".join("%s×%d" % (lb, len(clocks[lb])) for lb, _ in ROLES_CLOCK))

    # ⑧ 模型可用性（thinking=0 ＋ flash ⇒ 服务端必拒）
    deaf = ["%s（%s）" % ((a["name"] or "")[:26], a["model_id"]) for lst in clocks.values() for a in lst
            if not a["model_is_thinking"]
            and any(h in (a["model_id"] or "").lower() for h in FLASH_HINT)]
    if deaf:
        rec("model", "C", "fail", "周期钟会被服务端拒（静默失效）",
            "模型不支持关思考却传 thinking=0：%s" % "、".join(deaf))
    else:
        rec("model", "C", "ok", "周期钟模型可用", "")

    # ⑨ cwds 归属同形（错一字面 ⇒ 裂组且自我强化）
    off = _cwds_nearmiss(autos, ws_norm)
    if off:
        rec("cwd", "C", "fail", "`cwds` 与本工作区**近失配**（差一个字符就裂成两组）",
            "；".join(off[:3]))
    else:
        rec("cwd", "C", "ok", "`cwds` 归属同形", "本工作区的排期都写在同一条路径上")

    # ⑩ 投递心跳
    hbs = [WS + "/.workbuddy/collab/logs/supervise-heartbeat.json",
           WS + "/.workbuddy/collab/supervise-heartbeat.json"]
    age = None
    for h in hbs:
        if os.path.isfile(h):
            age = time.time() - os.path.getmtime(h)
            break
    if age is None:
        rec("deliver", "C", "warn", "投递（常驻）未见心跳",
            "⛔ 不等于它一定没跑；载体＝专用容器会话")
    elif age > 900:
        rec("deliver", "C", "fail", "投递心跳陈旧 ⇒ 常驻可能已掉线", "最后心跳 %.0f 分钟前" % (age / 60))
    else:
        rec("deliver", "C", "ok", "投递心跳新鲜", "%.0f 秒前" % age)

    # ⑪ 活会话
    live = {lb: [s for s in sess if (s[1] or "").startswith(pre) and (s[2] or "") == "working"]
            for lb, pre in ROLES_LIVE}
    dead = [lb for lb, _ in ROLES_LIVE if not live.get(lb)]
    if dead:
        rec("live", "C", "warn", "此刻无活会话（按需创建属正常；主会话开工阶段须建齐）", "、".join(dead))
    else:
        rec("live", "C", "ok", "三类会话均有活的",
            "；".join("%s×%d" % (lb, len(live[lb])) for lb, _ in ROLES_LIVE))

    # ⑫ 死排期：`once` 已无下次触发、却**从未运行过** ⇒ 活静默消失
    try:
        ran = {r[0] for r in c.execute(
            "select distinct automation_id from automation_runs limit 5000")}
    except Exception:
        ran = None
    never, consumed, unver, in_grace = _once_zombie(mine, ran)
    if never:
        rec("zombie", "C", "fail", "一次性排期**从未运行就失效**（那条活会静默消失）",
            "共 %d 条：%s" % (len(never), "、".join(never[:4])))
    elif unver:
        rec("zombie", "C", "warn", "无法核对一次性排期是否运行过",
            "读不到 `automation_runs`（⛔ 不等于它们没跑）；本线 %d 条 once＋无下次触发" % unver)
    else:
        rec("zombie", "C", "ok", "无「从未运行」的一次性排期",
            "本线 %d 条已跑完的一次性排期（正常痕迹，不计问题）%s"
            % (consumed, ("；%d 条刚到点、仍在补跑窗口内（⛔ 不是哑火）" % in_grace) if in_grace else ""))

# ── 收口 ───────────────────────────────────────────────────────────────
def finish():
    fails = [c for c in CHECKS if c["level"] == "fail"]
    warns = [c for c in CHECKS if c["level"] == "warn"]
    verdict = "fail" if fails else ("warn" if warns else "ok")
    summary = {"ok": "会话规则机制齐备且生效",
               "warn": "配置在册，运行未齐（见 warn 项）",
               "fail": "规则机制有硬缺口（见 fail 项）"}[verdict]
    state = {"ts": time.strftime("%Y-%m-%d %H:%M:%S"), "ts_epoch": int(time.time()),
             "ws": WS, "verdict": verdict, "summary": summary,
             "counts": {"fail": len(fails), "warn": len(warns),
                        "ok": len(CHECKS) - len(fails) - len(warns)},
             "checks": CHECKS}
    try:
        os.makedirs(os.path.dirname(OUT), exist_ok=True)
        tmp = OUT + ".tmp"
        with io.open(tmp, "w", encoding="utf-8", newline="\n") as f:
            f.write(json.dumps(state, ensure_ascii=False, indent=2))
        os.replace(tmp, OUT)          # 原子替换（本机文件锁会卡死 ⇒ 见 MEMORY）
    except Exception as e:
        state["write_error"] = str(e)[:160]

    if AS_JSON:
        sys.stdout.buffer.write((json.dumps(state, ensure_ascii=False, indent=2) + "\n").encode("utf-8"))
    else:
        icon = {"ok": "✅", "warn": "⚠️", "fail": "🔴"}[verdict]
        buf = ["%s [会话规则] %s（fail %d / warn %d / ok %d）"
               % (icon, summary, len(fails), len(warns), state["counts"]["ok"])]
        show = CHECKS if SHOW_ALL else [c for c in CHECKS if c["level"] != "ok"]
        for c in show:
            m = {"ok": "✓", "warn": "⚠", "fail": "✗"}[c["level"]]
            buf.append("      %s [%s] %s%s" % (m, c["group"], c["title"],
                                               (" —— " + c["detail"]) if c["detail"] else ""))
        if not SHOW_ALL and not show:
            buf.append("      （九项全过，明细见状态标记）")
        buf.append("      · 状态标记 → %s" % OUT)
        sys.stdout.buffer.write(("\n".join(buf) + "\n").encode("utf-8"))
    sys.stdout.flush()
    return 0 if verdict == "ok" else 1


def main():
    cmds, _ = check_hooks()
    if cmds:
        check_hook_skillnames(cmds)
        check_gate_logs()
    check_memory_pointers()
    check_snapshot_stale()
    check_orchestration()
    return finish()


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as e:
        try:
            sys.stdout.buffer.write(("⚠️ [会话规则] 体检自身异常（⛔ 不影响开工）：%s\n"
                                     % str(e)[:200]).encode("utf-8"))
        except Exception:
            pass
        sys.exit(0)