# -*- coding: utf-8 -*-
"""「本机协作」的**看板扩展** —— 🔴 本文件属于**本工作区**，⛔ 不属于技能。

规则（用户 2026-09-30 定）：
> **技能就是技能 程序就是程序，谁用产生的文件 放在他自己那里**

技能只提供**机制**（通用看板 ＋ 一个扩展挂点）；「链路前置」要看哪些组件、叫什么真名、
端到端怎么算通过 —— 都是**本项目的知识**，所以落在本文件里。

## 契约（由 `session-mechanism/scripts/board.py` 调用）
```python
def build(ws: str) -> dict
```
返回：
```python
{
  "title":       "链路前置",                 # 该板块标题
  "tag":         "组件状态",                 # 标题右侧的类别标签
  "chips":       [ {"label":…, "up":bool, "ok":str, "bad":str, "tip":str} ],
  "paragraphs":  [ {"text":…, "tone":"note"|"warn"} ],
  "tip":         "追加到该板块右上角 ? 里的项目侧说明",
}
```
⛔ 任何异常都必须被 board.py 吞掉并降级（看板照常，只是不显示这一块），所以本文件**自己也要**
保证不抛异常、且**只读**（⛔ 不写任何账本、⛔ 不发请求）。
"""
from __future__ import annotations

import glob
import json
import os
import re
import socket
import subprocess
import time

# [session-mechanism] roots.env 外置（由 install.py 生成；缺失则回落到按位置推导）
def _sm_load_roots():
    import os as _os
    _here = _os.path.dirname(_os.path.abspath(__file__))
    for _up in range(4):
        _p = _os.path.join(_here, *([".."] * _up), "roots.env")
        _p = _os.path.normpath(_p)
        if _os.path.isfile(_p):
            try:
                with open(_p, encoding="utf-8") as _f:
                    for _ln in _f:
                        _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()
# 🔴 2026-10-01 加 · **输出编码兜底**：脚本一旦被重定向（钩子/常驻/后台任务都会这么干），
#    Windows 本地编码（GBK）编不出 ⛔/✅/🔴 这类字符 ⇒ `print` 抛 UnicodeEncodeError
#    ⇒ 被顶层 handler 记成 `fatal`、**整轮失败**（实测：本包里连续 4 次 `fatal 'gbk' codec ...`）。
#    ⇒ 出口一律 UTF-8 ＋ errors="replace"（⛔ 不让"打不出字"升级成"程序死"）。
try:
    import sys as _sys
    _sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    _sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
    pass

# ── 本项目的组件定义（真名 · 系统-模块-功能名） ───────────────────────────
NODE_LOG_GLOB = os.environ.get("DSH_OVERLAY_LOG_GLOB") or ""
INGRESS_HOST, INGRESS_PORT = "127.0.0.1", 20090

# 🔴 **隐形执行体**登记册：走 WorkBuddy 网关 scheduled-tasks 建的任务**不进 automations 表**，
#    且 `durable:true` 的连网关自己的列表都列不出来 ⇒ 只认 id ⇒ 只能靠这份册子知道它还挂着。
SCHEDULES_REL = ".workbuddy/collab/gateway-schedules.json"


def _cron_zh(cron: str) -> str:
    """把 5 字段 cron 翻成人话。⚠️ 只覆盖本项目在用的两种形态，⛔ 不写通用 cron 解析器
    （写不全反而不如把原文显示出来）。认不出就**原样返回**，⛔ 不猜。"""
    c = str(cron or "").split()
    if len(c) != 5:
        return str(cron or "—")
    mi, hh, dom, mon, dow = c
    if mi.startswith("*/") and hh == dom == mon == dow == "*":
        return "每 %s 分钟" % mi[2:]
    if mi.isdigit() and hh.startswith("*/") and dom == mon == dow == "*":
        return "每 %s 小时的 %s 分" % (hh[2:], mi)
    return str(cron)


# 唤醒的**投递台账**里哪些 kind 算"一次真唤醒"（⚠️ 台账 append-only ⇒ 新旧名都要认）
# 🔴🔴 2026-10-02 **这两类 kind 都已随旧机制退役** —— ⛔ **别拿它代表"唤醒还在跑"**：
#    · `上报·唤醒` —— 由 `collabd.py::wake_round()` 写；该函数 **2026-09-29 已停用**；
#    · `监督程序·心跳` —— 由已停机的**守护进程**写。
#    实测台账（`wakeups.jsonl`，89 行）里命中的 42 条**最后一条停在 `2026-09-30T10:44:43`**；
#    而**现役**投递（`--tick`）写的是 **`上报·单条`**（`collabd.py` 的 `_deliver_str(..., "上报·单条", ...)`）。
#    ⇒ 故本常量只能读**历史痕迹**；⚠️ **"上次唤醒"的现役真读数在会话表**
#      （`sessions` 里 `[唤醒]` 那条的 `last_activity_at`，见 `_triggers()` 的 `detail2`）。
#    🔴 不许拿本常量去说"它今天已送 N 次" —— 那个数**结构性恒为 0**（＝又一处"在册 ≠ 生效"）。
WAKEUP_KINDS = ("上报·唤醒", "监督程序·心跳")
CONFIG_REL = ".workbuddy/collab/collabd.config.json"          # inbox 路径在里面


def _ago(epoch) -> str:
    try:
        m = int(max(0, (time.time() - float(epoch)) / 60.0))
    except Exception:
        return "—"
    return ("%d 分钟前" % m) if m < 60 else ("%.1f 小时前" % (m / 60.0))


def _stamp_age(ws: str, name: str):
    """某个戳文件的**年龄（分钟）**；没有/读不到 ⇒ None。

    🔴 为什么需要它：网关的 `durable` 定时任务**没有任何读接口能核实**——
    实测 `GET /api/v1/scheduled-tasks/{id}` → **404（无此路由）**、`?sessionId=` 只回 `{"tasks":[]}`
    （durable 的**永远列不出来**）。⇒ 想判"它还活着/还准点"，**只能让它自己留痕**。
    （2026-09-30 用户：「图上没有显示那条 bbaf9baf 已经不在了的实时状态」——
      当时框上写着「已登记」还带绿，是**假绿**，根因就是"登记册有记录"被当成了"它还活着"。）
    """
    try:
        p = os.path.join(ws, "tmp", "supervise-inbox", name)
        if not os.path.isfile(p):
            return None
        return int(max(0, (time.time() - os.path.getmtime(p)) / 60.0))
    except Exception:
        return None


def _last_wakeup(ws: str) -> dict:
    """🔴 唤醒的**真实执行痕迹** —— 读投递台账 `wakeups.jsonl`（每次投递都记 `ts` ＋ `http`）。

    ⚠️⚠️ **诚实边界（必须写进 tip，⛔ 不许包装成"运行中"）**：唤醒**只在三条前置同时成立时才投递**
    （🔴 2026-10-01 用户口径：**主会话与协作会话都空闲** ＋ **队列没有待反馈** ＋ **需求仍未完成**）
    ⇒ **两次投递之间是静默，不等于它没在跑**
    （我在干活的时候它就一直不投）。所以这一行给的是「上次**真的唤醒成功**在什么时候」——
    **真读数**，但 ⛔ 不是"它是否活着"的判据。
    """
    out = {"ts": "", "epoch": None, "http": None, "n_today": 0}
    try:
        with open(os.path.join(ws, CONFIG_REL), "r", encoding="utf-8") as f:
            inbox = json.load(f).get("inbox") or "tmp/supervise-inbox"
        rows = []
        with open(os.path.join(ws, inbox, "wakeups.jsonl"), "r", encoding="utf-8") as f:
            for ln in f:
                ln = ln.strip()
                if not ln:
                    continue
                try:
                    rows.append(json.loads(ln))
                except Exception:
                    pass
        hits = [r for r in rows if str(r.get("kind") or "") in WAKEUP_KINDS]
        if not hits:
            return out
        last = hits[-1]
        out["ts"] = str(last.get("ts") or "")
        out["epoch"] = last.get("epoch")
        out["http"] = last.get("http")
        today = time.strftime("%Y-%m-%d")
        out["n_today"] = sum(1 for r in hits if str(r.get("ts") or "").startswith(today))
    except Exception:
        pass
    return out


def _triggers_gw(ws: str) -> list:
    """⚠️ **旧通道**（WorkBuddy 网关 scheduled-tasks 册子 `gateway-schedules.json`）的读数。

    🔴 2026-10-01 起**不再作主读数** —— 桌面端调度器被产品关掉（`CODEBUDDY_DISABLE_CRON=1`），
    这份册子永远是 `tasks:0` ＋ 一堆历史 ⇒ 拿它当主源只能得出"通道已停用"这一句死结论，
    ⛔ 看不见"已经换成自动化排期了"这个事实。现役读数改由下面的 `_triggers()` 走 `automations` 表。
    本函数保留下来只为**解释这一段来历**（原注释原文如下），⛔ 不删。

    🔴 2026-10-02 补一句护栏：**本函数已无任何调用方**（实测
    `grep -n "_triggers_gw" board_ext.py` ⇒ 只有定义处、没有 `_triggers_gw(...)` 调用；
    现役入口是 `"triggers": _triggers(ws)`）⇒ ⛔ **改这里的文案／字段不会有任何效果**
    （包括下面那个 `"edge"` —— 它和 `_triggers()` 里那个同款，都已无人消费）。
    留着只为"当时的原注释"这句话本身，⛔ 别把它当成能生效的配置。

    🔴 外部**触发源**（时间驱动）—— 看板协作架构图 R5 行的右半。

    ⛔ 与「钩子子进程」（**事件驱动**）**不是一回事**，这也是本线最容易被混的一处：
      · 钩子子进程：宿主在钩子时机起它，跑完即退 ⇒ **整机没活动时它不会响**；
      · ~~心跳~~ **唤醒（时间驱动）**：网关注册的定时任务 ⇒ 补的正是"没人动时谁把 idle 的主会话叫起来"。
        🔴 **2026-10-02 标注（⛔ 不是口径变更）**：① 本函数 `_triggers_gw()` **已无调用方** ⇒ 这段只作历史；
        ② 「**用定时任务补时钟**」的方案 **2026-10-01 已废弃** —— 定案＝**常驻投递进程本身就是唤醒时钟**。

    ⚠️ 诚实边界：网关 `durable` 任务**列不出来**、也读不到 `last-run` ⇒
       这里**只能证明"登记还在"，⛔ 不能证明"还在跑"**。tip 里必须把这句话写出来，
       ⛔ 不许显示一个假的"运行中"（假绿比没有读数更糟）。
    """
    items = []
    try:
        with open(os.path.join(ws, SCHEDULES_REL), "r", encoding="utf-8") as f:
            doc = json.load(f)
        lw = _last_wakeup(ws)                     # 🔴 真实执行痕迹（投递台账）
        # 🔴 2026-09-30 回归修复 —— 用户原话：「**看板 唤醒节点都没了**」。
        #    根因：本函数只遍历**现役** `tasks`；现役一被清空，这一格就**整格消失**。
        #    ⇒ 判据修正：「唤醒」是架构图上的**固定占位**（表示"这台机器上到底有没有叫醒器"），
        #      ⛔ 没有现役 ≠ 这个概念不存在 ⇒ 回落**最近一条历史**（再没有才用匿名占位），
        #      并把"通道已停用"**显式写在 detail 里** —— ⛔ 不许悄悄消失，也⛔ 不许假绿。
        live = [(k, v) for k, v in (doc.get("tasks") or {}).items() if isinstance(v, dict)]
        has_live = bool(live)
        if not live:
            hist = [(k, v) for k, v in (doc.get("_历史") or {}).items() if isinstance(v, dict)]
            live = hist[-1:] or [("—", {"name": "唤醒", "cron": "—"})]
        for tid, t in live:
            cron = str(t.get("cron") or "")
            # 🔴 2026-09-30 用户定名：这一格图上叫 **「唤醒」**（原来叫"心跳"）。
            #    网关那条任务的**注册名**原样留在 `tip` 里（可回溯），⛔ 不在图上出现。
            nm = "唤醒"
            # ⚠️ 用户 2026-09-30：「把心跳 图框做小一点 放到**主会话左边**就行，本来就是主会话创建的」
            #    🔴 **位置这半句已被推翻** —— 用户 2026-10-01「**唤醒会话 和 跟进会话 单独放一行**」⇒
            #       现行位置＝**主会话下面那一行（`RW`）的左格**（布局常量见 `assets/board.html`）。
            #       这条只留**尺寸**依据（＝「做小一点」，故 `TBW=TW`），⛔ 别再照它摆位置。
            #    v2：「**里面是每5分钟核对目标状态**」⇒ detail 用这句。
            #    v3：「把唤醒的**真实执行情况 时间**也展示到架构图上」⇒ 第 3 行改**真读数**
            #        （读投递台账的**上次实际投递**时刻），来历挪进 tip。
            # 🔴 两个**真痕迹**（⛔ 都不是"登记册里有记录"那种推断）：
            #    ① `_wake.stamp`＝**cron 自己写的触发戳**（prompt 第 0 步要求它写）
            #    ② 投递台账＝**真投出去才记**
            sw = _stamp_age(ws, "_wake.stamp")
            lwmin = (int(max(0, (time.time() - float(lw["epoch"])) / 60.0)) if lw.get("epoch") else None)
            # ⚠️ 没有现役时**强制不绿** —— ⛔ 别让一个已停用的通道看起来还活着
            fresh = bool(has_live) and ((sw is not None and sw < 15) or (lwmin is not None and lwmin < 15))
            items.append({
                "name": nm,
                # 🔴 有现役 ⇒ 照旧写"多久核对一次"；无现役 ⇒ **直说通道停用**
                #    （2026-09-30：网关定时任务在桌面端被产品关掉，从未触发过一次）
                "detail": (("%s核对需求完成状态" % _cron_zh(cron)) if has_live
                           else "通道已停用 · 待换自动化"),
                # ⚠️ 用户 2026-09-30：「从没见过触发戳 这句话很难懂」⇒ 一律说大白话：
                #    「上次唤醒」＝它上一次真的响过；没响过就直说「无记录」，⛔ 不造词。
                "detail2": ("上次唤醒：" + ("%d 分钟前" % sw if sw is not None else "无记录")),
                "label": ("上次上报：" + ("%d 分钟前" % lwmin if lwmin is not None else "无记录")
                          + " · " + str(tid)[:8]),
                # 🔴 **颜色按"最近一次真痕迹"判** —— ⛔ 不再因为"登记册里有记录"就给绿（那是假绿）
                "up": bool(fresh),
                # 🔴 连线上写的动作是 **「唤醒」**（用户 2026-09-30：「唤醒和主会话 强调的是
                #    **唤醒的动作**，不是创建的动作」）⇒ 箭头由本框指向主会话。
                "edge": "唤醒",
                # ⚠️ 这一段进的是 SVG `<title>`，技能侧对它做 `esc()`，且 SVG title **本就只认纯文本**
                #    ⇒ ⛔ 不能用 `<br>`/`<b>`（会原样显示成标签）。用「；」分行。
                "tip": (("🔴 通道已停用：网关定时任务在**桌面端**被产品关闭"
                         "（宿主拼子进程环境时硬编码 CODEBUDDY_DISABLE_CRON=1 ⇒ 调度器不启动，"
                         "durable 写入还退化成静默空操作）⇒ 它**一次都没触发过**；"
                         "本格保留只表示「架构上还有没有叫醒器」，⛔ 不是在说它还活着；"
                         "正解＝换成自动化（循环最小 1 小时），待用户拍板。；" if not has_live else "")
                        + "%s；cron %s；%s；登记于 %s；上报到主会话 %s；来历：主会话创建的定时任务；"
                        "「上次唤醒」＝它自己每次响的时候写下的时间（写在 _wake.stamp 里），所以这是真读数；"
                        "「上次上报」＝它真的把话送到主会话的那一次（放在 wakeups.jsonl 台账里，%s http=%s，今天已送 %d 次）；"
                        "⚠️ 它只在三个条件同时成立时才出声（主会话与任务会话都空闲 ＋ 队列没有待反馈 ＋ 需求仍未完成），"
                        "所以两次之间有间隔是正常的；"
                        "⚠️ 但更要紧的是另一面：网关那条定时任务查不到状态"
                        "（按 id 查是 404，列表里也永远不显示）⇒ 所以⛔ 别信「登记册里记着就还在跑」，"
                        "只看上面两个时间；超过 15 分钟两个都没动，那一格就会变黄，意思就是「它没在响」。"
                        % (str(t.get("name") or tid), cron or "—",
                           "循环" if t.get("recurring") else "一次性",
                           t.get("created") or "—",
                           str(t.get("sessionId") or "")[:8],
                           lw.get("ts") or "—", lw.get("http"), lw.get("n_today") or 0)),
            })
    except Exception:
        return []
    return items


# ─────────────────────────────────────────────────────────────────────────────
# 🔴 「唤醒会话」的**真实运行状态** —— 用户 2026-10-01 两句话（**后一句覆盖前一句**）：
#    ① 「**唤醒机制 改为 唤醒定时任务**，要实时展示运行状态，包括图框样式要能体现」
#    ② 🔴「**唤醒会话，本质还是会话**」⇒ ⛔ 它不是调度配置 ⇒ **读数必须来自会话表**
#       （`sessions.status / last_activity_at`，⛔ 不再是 `automations` 排期表）。
#      ⛔ 两句共同点：「**不看册子里记着，只看真痕迹**」（本线已为「假绿」栽过跟头）。
#    ⚠️ **只读**：`busy_timeout` ＋ 只 `SELECT`。活库是 WAL 三件套、app 在线持有
#       ⇒ ⛔ 不写、⛔ 不拷文件、⛔ 不 `docker exec`。（见 MEMORY 活库改动铁律）
# ─────────────────────────────────────────────────────────────────────────────
CODEBUDDY_CFG = os.environ.get("CODEBUDDY_CONFIG_DIR") or os.path.expanduser("~/.workbuddy")
AUTOMATIONS_DB = os.path.join(CODEBUDDY_CFG, "workbuddy.db")


def _rrule_zh(rr: str) -> str:
    """rrule 翻人话。⚠️ 只覆盖本项目在用的几种形态（⛔ 不写通用解析器；认不出就**原样返回**）。"""
    s = str(rr or "").strip()
    if not s:
        return "一次性"
    u = {}
    for kv in s.split(";"):
        if "=" in kv:
            k, v = kv.split("=", 1)
            u[k.strip().upper()] = v.strip()
    f, ivl = u.get("FREQ", ""), u.get("INTERVAL", "1")
    hh, mm = u.get("BYHOUR", ""), u.get("BYMINUTE", "00")
    if f == "HOURLY":
        return "每小时" if ivl == "1" else ("每 %s 小时" % ivl)
    if f == "DAILY":
        if hh and "," not in hh:
            return "每天 %s:%s" % (hh.zfill(2), mm.zfill(2))
        if hh:
            return "每天 %s 点" % hh
        return "每天" if ivl == "1" else ("每 %s 天" % ivl)
    if f == "WEEKLY":
        wd = "/".join([x for x in (u.get("BYDAY", "") or "").split(",") if x])
        return "每周%s%s" % (wd, (" %s:%s" % (hh.zfill(2), mm.zfill(2))) if hh else "")
    if f == "MONTHLY":
        return "每月 %s 日%s" % (u.get("BYMONTHDAY", ""),
                                (" %s:%s" % (hh.zfill(2), mm.zfill(2))) if hh else "")
    return s


def _ts_zh(ms) -> str:
    """毫秒 epoch → `MM-DD HH:MM`；给不出就「—」。⚠️ 表里存的是**毫秒**，⛔ 别当秒。"""
    try:
        return time.strftime("%m-%d %H:%M", time.localtime(float(ms) / 1000.0))
    except Exception:
        return "—"


def _ago_ms(ms):
    """毫秒 epoch → 「N 分钟前／N.N 小时前」；给不出就 None。"""
    try:
        m = int(max(0, (time.time() - float(ms) / 1000.0) / 60.0))
    except Exception:
        return None
    return ("%d 分钟前" % m) if m < 60 else ("%.1f 小时前" % (m / 60.0))


def _wakeup_autos(ws: str):
    """读 `automations` 表里**属于本工作区**的唤醒类排期。

    返回 `list[dict]`；**读不到**（库不在／被锁／无表）⇒ 返回 `None`。
    ⚠️ `[]`（确无）与 `None`（读不到）**必须可分** —— 否则 tip 会把「查不到」说成「没有」，
       这正是本线反复栽的「假绿／假红」那类错。
    """
    try:
        import sqlite3
        if not os.path.isfile(AUTOMATIONS_DB):
            return None
        wn = os.path.basename(os.path.normpath(ws)).lower()
        con = sqlite3.connect(AUTOMATIONS_DB, timeout=4)
        try:
            con.execute("PRAGMA busy_timeout=4000")
            cur = con.execute("SELECT id,name,status,schedule_type,rrule,next_run_at,last_run_at,cwds,"
                              "valid_until FROM automations")
            now_s = time.strftime("%Y-%m-%dT%H:%M")
            rec, once_n = [], 0
            for rid, nm, stt, styp, rr, nx, ls, cw, vu in cur:
                nm = str(nm or "")
                if ("唤醒" not in nm) and ("心跳" not in nm):
                    continue
                if wn and wn not in str(cw or "").lower():
                    continue
                # ⛔ **只认周期性排期** —— 「定时任务」的本质是"到点自己响"。
                #    一次性派棒（如「接续 · 机制线复核（唤醒回路实战）」「主控 · 机制线（查唤醒为什么断）」）
                #    名字里带着「唤醒」二字，但它们是**派活**，⛔ 不是叫醒器。
                #    ⚠️ 2026-10-01 实测踩到：不加这条，11 条里 8 条是一次性 ⇒ 被判成 `running` ＝ **假绿**。
                if str(styp or "") != "recurring":
                    once_n += 1
                    continue
                rec.append({"id": str(rid), "name": nm, "status": str(stt or ""),
                            "type": str(styp or ""), "rrule": str(rr or ""),
                            "next": nx, "last": ls,
                            # ⚠️ `valid_until` 是 **TEXT（ISO 分钟精度）**，⛔ 不是毫秒数 ⇒ 按字符串比
                            "expired": bool(vu) and (str(vu) < now_s)})
            return {"rec": rec, "once_n": once_n}
        finally:
            con.close()
    except Exception:
        return None


def _waker_session(ws: str):
    """🔴 找**唤醒会话**（`sessions.title` / `custom_title` 前缀 `[唤醒]`）并读**它自己的状态**。

    用户 2026-10-01 定性（**后一句覆盖前一句**）：
      · 「**唤醒会话，本质还是会话**」⇒ 这一格 ⛔ 不是"调度配置"，⛔ 不再读 `automations` 排期表；
        读的是**这条会话自己**：在不在、在跑还是停着、最近动过没。
      · 🔴「**唤醒会话 是独立会话，不要在主会话上处理，是随着需求确定时创建的**」
        ⇒ (a) **独立会话** —— 它自己一条，⛔ 不是主会话的一个"职能"；
           (b) ⛔ **不在主会话上处理** —— 叫醒的活儿由它自己干，⛔ 别挂在主会话身上；
           (c) **随需求确定时创建** —— 需求确定才建它；没需求不建，⛔ 也不提前建。

    返回 `list[dict]`（最多 1 条）；**读不到**（库不在／被锁）⇒ `None`
    （⚠️ `[]`＝确无 与 `None`＝读不到 **必须可分** —— 本线反复栽在"把查不到说成没有"上）。
    """
    try:
        import sqlite3
        if not os.path.isfile(AUTOMATIONS_DB):
            return None
        wn = os.path.basename(os.path.normpath(ws)).lower()
        con = sqlite3.connect(AUTOMATIONS_DB, timeout=4)
        try:
            con.execute("PRAGMA busy_timeout=4000")
            cur = con.execute(
                "SELECT id,title,custom_title,status,cwd,updated_at,last_activity_at FROM sessions "
                "WHERE (title LIKE '[唤醒]%' OR custom_title LIKE '[唤醒]%') "
                "AND (deleted_at IS NULL OR deleted_at=0) "
                "ORDER BY updated_at DESC")
            out = []
            for sid, ti, ct, stt, cw, up_, la in cur:
                if wn and wn not in str(cw or "").replace("\\", "/").lower():
                    continue
                out.append({"id": str(sid or ""), "title": str(ct or ti or ""),
                            "status": str(stt or "").lower(),
                            "updated_at": up_, "last_activity_at": la})
            return out
        finally:
            con.close()
    except Exception:
        return None


def _triggers(ws: str) -> list:
    """🔴 主会话**左边**那一格 ＝ **唤醒会话**。用户 2026-10-01 三次定义（**后一句覆盖前一句**）：
    ① 「唤醒机制 改为 唤醒定时任务，要实时展示运行状态，包括图框样式要能体现」
    ② 🔴「**唤醒会话，本质还是会话**」⇒ ⛔ 不是调度配置 ⇒ 读**会话表**，⛔ 不读排期表。
    ③ 🔴「**唤醒会话 是独立会话，不要在主会话上处理，是随着需求确定时创建的**」⇒ 三条硬定性：
       **(a) 独立会话**（自己一条，⛔ 不是主会话的职能）；**(b) ⛔ 不在主会话上处理**（叫醒的活儿它自己干）；
       **(c) 随需求确定时创建**（需求确定才建它 ⇒ 没需求不建，也⛔ 不提前建）。

    状态三态（`state` 字段，画图侧照它换样式 ⇒ ⛔ 看图就知道它活着没）：
      · `running` —— 会话 `status='working'` ⇒ **绿**（正醒着干活）；
      · `paused`  —— 会话在、但没在跑（`idle` 等脉冲到点）⇒ **黄**（在，但此刻没动）；
      · `none`    —— 这条会话**还没建** ⇒ **灰**（架构上有它，实际还没创建 — 按定性要等需求确定）。

    ⛔ 绝不因「册子里记着」就给绿 —— 只看**会话表里它自己的状态**（也不看排期表）。
    """
    sess = _waker_session(ws)          # None ＝ 读不到；否则 list[dict]（最多 1 条）
    lst = sess or []
    lw = _last_wakeup(ws)

    if lst:
        s = lst[0]
        live = (s["status"] == "working")
        state, up = (("running" if live else "paused"), live)
        detail = ("会话在跑（脉冲到点了）" if live
                  else ("会话空闲 · 等脉冲到点" if s["status"] == "idle"
                        else "会话状态：" + (s["status"] or "未知")))
        detail2 = "最近活动：" + (_ago_ms(s.get("last_activity_at") or s.get("updated_at"))
                                  or "无记录")
        label = "id %s · %s" % (s["id"][:8], s["status"] or "状态未知")
    else:
        state, up = ("paused" if sess is None else "none"), False
        detail = ("读不到会话表（数据库未就绪）" if sess is None
                  else "还没建 · 按定性要等需求确定时才建")
        # 🔴 2026-10-01 用户改文案：「上次脉冲响过」⇒「**上次唤醒时间**」（说人话，⛔ 别讲机制黑话）
        # ⚠️ 2026-10-02 补一句：本分支只在**一条 `[唤醒]` 会话都没有**时才走到，而 `lw` 读的是
        #    **已退役**的两类 kind（见 `WAKEUP_KINDS` 上方的说明）⇒ 这个时间**只能当历史痕迹**看，
        #    ⛔ 不能据此说"机制在跑"。真读数永远是会话表那条（有会话时走上面的分支）。
        detail2 = "上次唤醒时间：" + (_ago(lw["epoch"]) if lw.get("epoch") else "无记录")
        label = "架构上还有没有这条会话"

    tips = ["这一格 ＝ **唤醒会话** —— 用户 2026-10-01 三条定性："
            "① 「**唤醒脉冲会话，本质还是会话**」（⚠️ 用户原话引用保留；2026-10-01 改名后现名＝**唤醒会话**）⇒ **它就是一条会话**；"
            "② 「**唤醒会话 是独立会话，不要在主会话上处理**」⇒ 自己一条，⛔ 不是主会话的职能；"
            "③ 「**是随着需求确定时创建的**」⇒ 需求确定才建，没需求不建、⛔ 也不提前建。"
            "位置在**主会话下面那一行**（左格，右格是跟进会话）—— 用户 2026-10-01：「**唤醒会话和 跟进会话 单独放一行**」；⛔ 不进下方任务会话排；⛔ 它也不是调度配置。"]
    if sess is None:
        tips.append("⚠️ 读不到会话表（数据库未就绪／被占用）⇒ **状态未知**，⛔ 别当成「没有」")
    elif not lst:
        tips.append("⛔ 本工作区**还没建这条会话** —— 按定性它**要等需求确定时才建**；"
                    "在那之前，没有任何东西会在没人动的时候把主会话叫起来")
    else:
        for s in lst:
            tips.append("· %s（%s；id %s；最近活动 %s）"
                        % (s["title"] or "（无标题）", s["status"] or "—", s["id"][:8],
                           _ago_ms(s.get("last_activity_at") or s.get("updated_at")) or "无记录"))
    tips.append("🔴 **它什么时候才出声**（三条件同时成立）：① 主会话与任务会话**都空闲**"
                " ② **队列没有待反馈** ③ **需求仍未完成**；"
                "需求**有阻碍**或**已完成** ⇒ 唤醒任务应**暂停**（⛔ 别在卡住时重复叫）。")
    tips.append("⚠️ 判据看**这条会话自己**：在 `working` ⇒ 绿（它正醒着干活）；存在但不在跑 ⇒ 黄；"
                "找不到 ⇒ 灰。")
    # 🔴 2026-10-02 **订正由来的那半句**（原文：「⛔ 不再数排期条数 —— 唤醒已从『排期开新会话』
    #    改成『脉冲复用同一条会话』」）—— 🔴 **后半句与实测相反，已废**：
    #    只读宿主库实测（`automations` ＋ `automation_runs` ＋ `sessions`）＝
    #      排期 `[唤醒]-…-脉冲` 是 `recurring / FREQ=HOURLY;INTERVAL=1`，**每跳真的开一条新会话**
    #      （同标题的会话每小时新增一条、id 各不相同：…05:06:24／06:08:00…），跑完即 `completed`。
    #    ⚠️ "**不再数排期条数**"这个做法**仍然对**（本格读的是会话、不是排期）⇒ 做法保留，**理由重述**：
    #       不是"改成复用同一条了"，而是"这一格问的是『这条会话此刻活着没』，排期条数答不了这个问题"。
    tips.append("⚠️ 本格读的是**一条会话**（`sessions` 里标题带 `[唤醒]` 的那条），⛔ 不读数排期条数。"
                "🔴 但**触发它的现在是一条宿主排期、且每跳开一条新会话** ⇒ 框里那条读数永远是"
                "「**最近一跳**」，会话跑完就变黄／灰。"
                "⚠️⚠️ **这不是定案形态** —— 定案是「**常驻投递进程就是唤醒时钟**」（`--supervise` · "
                "0 token · 0 会话），**⛔ 不用排期**（2026-10-01 你定的「定时任务的方案已经废弃了」）；"
                "现在由排期代偿，**只是因为常驻不在**（全机无 `--supervise`）。"
                "⇒ 「每跳开新会话」是**代偿现象的读数**，⛔ 不是「机制本来就该这样」。")
    # 🔴 2026-10-02 **订正**：原文是「⚠️ **诚实边界**：「**下次几点到点**」⛔ 读不到 —— 脉冲是会话
    #    自己起的**后台任务**，没有公开接口」—— 🔴 **这句两半都错**：
    #      ① 触发源**不是**"会话自己起的后台任务"，是**宿主排期**（`automations` 表里那一条）；
    #      ② 「下次几点到点」**读得到** —— `automations.next_run_at` 就是它（实测读到 `10-02 07:10:34`）。
    #    ⚠️ 但本格**目前还没读它**（＝已知缺口，已登记）⇒ 文本改成"读得到、只是本格还没读"，
    #       ⛔ 不再说"读不到"（那是把**本格没做**说成**产品没有** ⇒ 下一个"在册 ≠ 生效"的坑）。
    # 🔴 2026-10-02 **第三处订正（读数口径）**：上面这两句原来把 `_last_wakeup()` 的回读说成
    #    "**上次真的响过**／**今天已送 N 次**" —— 🔴 **那个数不可信、原句已废**：
    #    `WAKEUP_KINDS` 的两类 kind 都随旧机制退役（台账里最后一条停在 **2026-09-30T10:44:43**）
    #    ⇒ 照原样写就是"看板说唤醒两天没响过"，**而事实是它每小时都在响**
    #    （`sessions` 里 `[唤醒]-…` 每小时新增一条：…03:55:52／05:06:24／06:08:00…）。
    #    ⇒ 改法：**这个数不再当读数用**，只在文案里点出"台账口径已过时"，并指向真读数。
    tips.append("⚠️ 「**上次真的响过是什么时候**」⛔ **不看投递台账** —— 台账里那两类「唤醒」记录"
                "（`上报·唤醒`／`监督程序·心跳`）**已随旧机制退役**，最后一条停在 %s；"
                "拿它当读数就会得出「好几天没响过」的**假结论**。"
                "✅ **真读数 ＝ 上面那条会话的「最近活动」**（`sessions` 表，机制每跳开一条新会话）。"
                "⚠️ 触发源是**宿主排期**（⛔ 不是会话自己起的后台任务）⇒ 「**下次几点到点**」其实"
                "**读得到**（排期表里就有），只是**本格还没读它** —— 属**已知缺口**，⛔ 不编一个数出来"
                % (lw.get("ts") or "无记录"))

    return [{
        "name": "唤醒",
        # 🔴 名字由**项目侧**给（用户 2026-10-01 定名），技能侧只负责画 ⇒ ⛔ 技能里不出现这个词。
        # 🔴 2026-10-01 23:0x 用户改**简称**：「**唤醒会话 改为  唤醒，跟进会话 改为 跟进  字体和 唤醒一样大小**」
        #    ⇒ 这一格标签用**简称「唤醒」**（全称仍是「唤醒会话」），与右格「跟进」**同字号**。
        #    ⚠️ 图上简称、正文全称 —— 两者是**同一个东西**，⛔ 别当成两个角色（`?` 里已写明对照）。
        "detail": detail,
        "detail2": detail2,
        "label": label,
        "up": bool(up),
        "state": state,                        # 🆕 给样式用：running／paused／none
        # 🔴 2026-10-02 **删掉死字段 `edge`**（原为 `"edge": "唤醒"`，注释写"箭头指主会话"）：
        #    ⚠️ 它**已经没有任何消费方** —— `assets/board.html` 里那句 `if(t.edge) lbl(...)`
        #       在 2026-10-02 改连线（唤醒 → 常驻程序）时被换成了**固定文案**
        #       （`唤醒 · --tick 推一次`）⇒ 这个键从此没人读。
        #    🔴 判据（可复跑）：`grep -rn "\.edge" --include=*.js --include=*.py --include=*.html`
        #       ⇒ **零命中**（只剩 CSS 里的 `.edge{}` 类名）。
        #    🔴 留着它就是下一个「**在册 ≠ 生效**」的坑：字段还在册、改它却毫无反应，
        #       而且它的注释还是**已被推翻的旧口径**（"箭头指主会话"）⇒ 删掉、⛔ 不留。
        # ⚠️ 这一段进的是 SVG `<title>` ⇒ 纯文本（⛔ 不能用 `<br>`/`<b>`），用「；」分行。
        "tip": "；".join(tips),
    }]


# 监督守护的停因（属本项目的历史事实，故写在项目侧）
GUARD_STOP_REASON = (
    "2026-09-29 23:59 由主会话 `guard.py --stop` 主动止损（guard.log `23:59:01 收到停止标志`）："
    "守护当时是用「会话内后台任务」起的，任务归属发起会话，那个会话被判定一直挂着长跑任务，"
    "实测「一启动会话就卡消息输出」。根因是「有网关口令」与「不占会话」不可兼得；"
    "正解是拆两件：「发现」走会话外常驻（启动文件夹／独立窗口，但拿不到口令）、"
    "「投递」走宿主起的通道（钩子 `--tick` ／ 低频排期，宿主起的子进程天生有口令）。"
    "🔴 2026-10-01 用户定案（**覆盖上面那句「不再需要」**）：「**协作与投递一直运行（常驻）**」"
    "（理由「**可能不是所有队列都是钩子产生的**」）＋「**定时任务的方案已经废弃了**」"
    "⇒ 正解＝在**专用容器会话**里跑 `collabd.py --supervise`（stdout 全重定向；"
    "⛔ 不在「会干活的会话」里起），宿主钩子只作**补充**。"
    "⚠️ 因此**「停着」现在是一个真读数**（常驻该起没起），⛔ 不再是「没有启动／停止这回事」。"
)


def _listening(port: int) -> bool:
    """🔴 **端口是否真有进程在 LISTENING** —— ⛔ 不能只靠 TCP connect：

    2026-09-30 实测踩到：本机装了 **Proxifier**，它把**回环请求也代理**掉 ⇒
    端口**没在监听**时 `connect()` 依然成功（它本地替你把连接吃下了）⇒ **假阳性**。
    那一轮看板就显示「:20090 在监听」，而同刻 `netstat` 里**根本没有该端口**。

    ⇒ 判据改为：**以 `netstat -ano` 的 LISTENING 行为准**（拿不到 netstat 才回落到 connect 探测）。
    """
    try:
        out = subprocess.run(["netstat", "-ano"], capture_output=True, text=True,
                             timeout=6, errors="replace",
                             creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)).stdout or ""
        pat = re.compile(r"[:.]%d\s" % int(port))
        for ln in out.splitlines():
            if "LISTENING" in ln and pat.search(ln):
                return True
        return False
    except Exception:
        pass
    # 回落：拿不到 netstat（极少见）才用 connect —— ⚠️ 这时结论**可能假阳**，所以记进 tip
    s = socket.socket()
    s.settimeout(0.3)
    try:
        s.connect(("127.0.0.1", int(port)))
        return True
    except Exception:
        return False
    finally:
        try:
            s.close()
        except Exception:
            pass


def _relay_tail() -> dict:
    """读中继客户端自己写的状态行（⛔ 只读日志，⛔ 不发请求）。"""
    out = {"up": False, "state": "", "streams": None, "denied": None,
           "in_b": None, "out_b": None, "last_error": "", "up_s": None, "age_min": None}
    try:
        fs = sorted(glob.glob(NODE_LOG_GLOB), key=os.path.getmtime)
        if not fs:
            return out
        newest = fs[-1]
        out["age_min"] = round((time.time() - os.path.getmtime(newest)) / 60.0, 1)
        with open(newest, "rb") as f:
            f.seek(max(0, os.path.getsize(newest) - 8192))
            lines = f.read().decode("utf-8", "replace").strip().splitlines()
        row = ""
        for ln in reversed(lines):
            if "[relay-client]" in ln and "state=" in ln:
                row = ln
                break
        if not row:
            return out
        for key, pat in (("state", r"state=(\w+)"), ("streams", r"streams=(\d+)"),
                         ("denied", r"denied=(\d+)"), ("in_b", r"\bin=(\d+)B"),
                         ("out_b", r"\bout=(\d+)B")):
            m = re.search(pat, row)
            if m:
                out[key] = m.group(1) if key == "state" else int(m.group(1))
        m = re.search(r"\(for (\d+)ms\)", row)
        if m:
            out["up_s"] = round(int(m.group(1)) / 1000.0, 1)
        m = re.search(r'lastError="([^"]*)"', row)
        if m:
            out["last_error"] = m.group(1)
        out["up"] = (out["state"] == "up")
    except Exception:
        pass
    return out


def _fmt_b(n):
    if not isinstance(n, int):
        return "—"
    return ("%d B" % n) if n < 1024 else (("%.1f KB" % (n / 1024.0)) if n < 1048576
                                          else ("%.1f MB" % (n / 1048576.0)))


def build(ws: str) -> dict:
    ingress_up = _listening(INGRESS_PORT)
    node = _relay_tail()
    e2e = isinstance(node.get("streams"), int) and node["streams"] > 0

    chips = [
        {"label": "dsh-plugin-ai1net · 设备接入 · 本地中转 :%d" % INGRESS_PORT,
         "up": ingress_up, "ok": "在监听", "bad": "未监听",
         "tip": "看 127.0.0.1:%d 这个口有没有人在听（只读探测，不发请求；⚠️ 本机装了 Proxifier 会代管本机连接 ⇒ 不能只看能不能连上）" % INGRESS_PORT},
        {"label": "dsh-client · 覆盖网络节点 · 中继客户端",
         "up": bool(node.get("up")), "ok": "已注册", "bad": "未注册",
         "tip": "中继客户端的原始读数：状态=%s · 转发过=%s · 被拒=%s · 收=%s · 发=%s · 已在线 %s 分钟%s"
                % (node.get("state") or "?", node.get("streams"), node.get("denied"),
                   _fmt_b(node.get("in_b")), _fmt_b(node.get("out_b")),
                   int((node.get("up_s") or 0) // 60),
                   (" · lastError=" + node["last_error"]) if node.get("last_error") else "")},
    ]

    paras = [{"text": "链路：手机 → 覆盖网络中继 → 本机「设备接入 · 本地中转」（127.0.0.1:%d）→ 桌面 DSH 会话"
                       % INGRESS_PORT, "tone": "note"}]
    if e2e:
        paras.append({"text": "中继转发过 %d 条连接，从手机到桌面真的通了。" % node["streams"], "tone": "note"})
    else:
        paras.append({"text": "两个组件都在线，链路还没通：中继客户端只是注册上了，转发计数=%s 说明还没有请求被真正转发过%s，端到端没验过。"
                              % (node.get("streams"), ("（denied=%s）" % node["denied"]) if node.get("denied") is not None else ""),
                      "tone": "warn"})
    if not (ingress_up and node.get("up")):
        paras.append({"text": "有组件没起来，平台入口会回 503。重起见 references/deploy.md §5b。", "tone": "warn"})

    # 监督守护停着 ⇒ 附停因（这一条属本项目）
    try:
        guard_stopped = os.path.isfile(os.path.join(ws, "tmp", "supervise-inbox", "guard.stop"))
    except Exception:
        guard_stopped = False
    if guard_stopped:
        # ⚠️ 原文写「监督守护已停 ⇒ 投递改由钩子…」—— 2026-09-30 用户指出这句在误导：
        #    「监督程序」已退役（它不是一个该常驻的进程），**没有"停不停"这回事**。
        #    ⇒ 改成只说**事实与代价**，⛔ 不再提那个名字。
        # 🔴 2026-10-01 再修（**假绿式文案**）：原句写「补这个缺口的正是…那格「心跳」」——
        #    ⛔ 两处错：① 那格当时的名字已不是「心跳」（现名「**唤醒会话**」）；
        #    ② 写这话时它 **3 条周期排期全部 PAUSED、`last_run_at` 全空（从未触发过）**
        #       ⇒ 说"缺口有人补"是**宣称一个当前不成立的能力**
        #       （同日 00:59 STALL.md 实测：「5.3 小时无成果 ＋ 未来 1 小时零排期 ⇒ 确定性静默」）。
        #    ⇒ 改成**先查这条会话的真实状态再说**（三态：在跑／在但没跑／还没建，⛔ 不许合并）。
        #    🔴 2026-10-01 用户再加一条定性：「**唤醒会话 是独立会话，不要在主会话上处理，
        #       是随着需求确定时创建的**」⇒ 下面"还没建"那一支要**说明这是正常的**：
        #       它按定性**要等需求确定时才建**，⛔ 不是"该有却没有"的故障。
        _wsx = _waker_session(ws)
        _live = [s for s in (_wsx or []) if s.get("status") == "working"]
        if _wsx is None:
            _w = "（那格的状态这会儿读不到，⛔ 别当成它在跑）"
        elif _live:
            _w = "补这个缺口的是主会话下面那一行左格「唤醒会话」—— 它现在正醒着干活。"
        elif _wsx:
            _w = ("⚠️ 主会话下面那一行左格「唤醒会话」在，但此刻没在跑"
                  "（现状由排期代偿到点叫醒；⚠️ 定案是**常驻投递直接叫醒**）"
                  "—— 这段空窗里没人发消息的话，整条链不会自己动。")
        else:
            _w = ("⚠️ 本该补这个缺口的是主会话下面那一行左格「唤醒会话」，但它**还没建**"
                  "（按定性它**随需求确定时才创建**）"
                  "—— 这个缺口眼下是敞开的：没人发消息，整条链就不会动。")
        paras.append({"text": "上报／投递按定案一直运行（常驻，`--supervise`）；宿主钩子只作补充"
                              "（事件驱动，跑完即退）⇒ 常驻没拉起时，整机完全没有活动就没人推队列。" + _w,
                      "tone": ("note" if _live else "warn")})

    return {
        "title": "链路前置",
        "tag": "组件状态",
        "chips": chips,
        "paragraphs": paras,
        "tip": ("组件名写成「系统 · 模块 · 功能名」，不用自造简称。<br>"
                "状态从组件自己留下的时间记录和日志里读；只做探测，不发请求。<br>"
                "两个组件都在线，链路也可能是断的。只有中继真的转发过，才算从手机到桌面通了。"),
        "guard_stop_reason": GUARD_STOP_REASON if guard_stopped else "",
        # 🔴 外部**触发源**（时间驱动）→ 让看板架构图画出**唤醒会话**那格。
        #    形状是**通用**的（技能侧只认这个形状，⛔ 不认识"心跳"这个名字）。
        "triggers": _triggers(ws),
    }