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

用户口径（2026-10-02 原话逐字）：
  「这个会话和协作会话的技能包 运行的第一件事 ，就应该是检查清楚
    所有会话规划是否配置完整且生效，然后标记一个状态」

为什么要它（真因，不是想当然）
──────────────────────────────
2026-10-02 实测：本线的排期**都在册**（`[唤醒]-…` / `[跟进]-…` 两条 recurring、
`model_is_thinking=1`、`cwds` 正确），**但三类会话一条活的都没有** ——
"配置在册" ≠ "机制在生效"。同一天早些时候还发现过更隐蔽的：
钩子注入文本指着 **09-28 就已合并退役**的技能名、常驻规则快照的生成脚本**写到幽灵目录**
⇒ 全都是"看着有配置、其实没生效"。所以**开工第一件事**必须先把这个状态问清楚，
而不是等它表现出来（那时已经卡住或被用户发现）。

检查六项（⛔ 每条都必须能真报出问题 —— 见 selftest 的红绿对照）：
  ① 周期钟    每类是否有 recurring 排期（缺 ⇒ 没人推 / 没人收）
  ② 钟会响吗  排期绑的模型能不能跑（`thinking=0` ＋ flash 系 ⇒ 服务端必拒，2026-10-02 实测）
  ③ 归属同形  `cwds` 与工作区路径**逐字**比（错一字面 ⇒ 裂组且**自我强化**）
  ④ 活会话    三类（唤醒 / 协作 / 跟进）当前有没有活的
  ⑤ 投递在跑  协作程序（投递轮）在不在跑 ＋ 心跳新鲜度
  ⑥ 死排期    `once` 且 `next_run_at` 为空 ⇒ 已消耗、**不会再触发**（静默消失的活）

产出（**标记一个状态**）：
  `<WS>/.workbuddy/collab/session-plan.json` —— 机器可读（看板 / 后续会话读它）
  ＋ stdout 一行摘要（给人扫）

退出码：0 = 齐备且生效 ｜ 1 = 有缺口（⚠️ **只标记、不拦开工**）
用法：python session-plan-check.py [--ws <工作区>] [--json] [--quiet]
"""
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")
OUT = WS + "/.workbuddy/collab/session-plan.json"   # ⛔ 不用 os.path.join（Windows 会出反斜杠，混排）
AS_JSON = "--json" in ARGV
QUIET = "--quiet" in ARGV

# 四类会话的标题前缀（口径 ⇒ skills/session-mechanism · 用户 2026-10-01 定「四类」）
ROLES = [("唤醒", "[唤醒]"), ("协作", "[协作]"), ("跟进", "[跟进]")]
# ⚠️ 只有「唤醒 / 跟进」要求**周期钟** —— 它们是"机制在跑"的两台钟（推 / 收）。
#    **协作会话是按需创建的**（用户 2026-10-01 口径「随需求确定时创建」）⇒ 要求它有 recurring
#    排期就是**假警报**（本条判据首版就是这么写错的：首次实跑报了「协作类周期钟缺失」，
#    而那是设计如此）。判据必须跟口径同源，⛔ 不按"四类应当对称"想当然。
ROLES_CLOCK = [("唤醒", "[唤醒]"), ("跟进", "[跟进]")]
# flash 系模型**不支持关闭思考** ⇒ `model_is_thinking=0` 时服务端 -32603（`automation-request-refused`）
FLASH_HINT = ("flash",)
# 排期"已终结"的时长上限：`next_run_at` 距现在超过这么久 ⇒ 判为"钟停了"
STALE_FUTURE_MS = 365 * 24 * 3600 * 1000

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

def _ro_conn():
    """活库铁律：只读打开 ＋ busy_timeout（⛔ 绝不写宿主库）。"""
    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 _now_ms():
    return int(time.time() * 1000)

def main():
    if not os.path.isfile(DB):
        rec("db", "fail", "读不到宿主库", "库不在：%s" % DB)
        return finish()
    try:
        c = _ro_conn()
    except Exception as e:
        rec("db", "fail", "打不开宿主库（只读）", str(e)[:160])
        return finish()

    # ── 取数：全部未删排期 ＋ 本工作区相关的会话 ─────────────────────────
    try:
        autos = list(c.execute(
            "select id,name,status,schedule_type,next_run_at,rrule,model_id,"
            "model_is_thinking,cwds,last_run_at from automations where deleted_at is null"))
    except Exception as e:
        rec("db", "fail", "读 automations 失败", str(e)[:160])
        return finish()
    try:
        sess = list(c.execute(
            "select id,title,status,cwd,updated_at from sessions "
            "where title like '[唤醒]%' or title like '[协作]%' or title like '[跟进]%'"))
    except Exception:
        sess = []

    ws_norm = _norm_cwd(WS)
    mine = [a for a in autos if any(_norm_cwd(x) == ws_norm
                                   for x in (_safe_list(a[8])))] or []

    # ── ① 周期钟：唤醒 / 跟进 两类必须有 recurring 排期 ────────────────────
    clocks = {}
    for a in autos:
        nm = a[1] or ""
        if a[3] != "recurring":
            continue
        if not any(_norm_cwd(x) == ws_norm for x in _safe_list(a[8])):
            continue
        for label, pre in ROLES_CLOCK:
            if nm.startswith(pre):
                clocks.setdefault(label, []).append(a)
    missing_clock = [lb for lb, _ in ROLES_CLOCK if not clocks.get(lb)]
    if missing_clock:
        rec("clock", "fail", "周期钟缺失",
            "这些类别没有任何 recurring 排期 ⇒ 没人在推 / 收：%s" % "、".join(missing_clock))
    else:
        rec("clock", "ok", "周期钟在册",
            "；".join("%s×%d" % (lb, len(clocks[lb])) for lb, _ in ROLES_CLOCK)
            + "（协作按需创建 ⇒ 不要求周期钟）")

    # ── ② 钟会响吗：模型可用性（thinking=0 ＋ flash ⇒ 必被拒） ─────────────
    deaf = []
    for lb, lst in clocks.items():
        for a in lst:
            mid = (a[6] or "").lower()
            think = a[7]
            if not think and any(h in mid for h in FLASH_HINT):
                deaf.append("%s（%s）" % (a[1][:28], a[6]))
    if deaf:
        rec("model", "fail", "钟会被服务端拒（静默失效）",
            "模型不支持关闭思考却传了 thinking=0 ⇒ 每触发必拒：%s" % "、".join(deaf))
    else:
        rec("model", "ok", "排期模型可用",
            "周期排期均未命中「flash ＋ 关思考」这一必拒组合")

    # ── ③ 归属同形：cwds 逐字比 ──────────────────────────────────────────
    off = []
    for a in autos:
        for x in _safe_list(a[8]):
            s = str(x or "")
            if not s.strip():
                continue
            if _norm_cwd(s) != s.replace("\\", "/").strip().lower():
                pass
            # 只报"本线的排期但字面不符"（别人的 cwds 不关我事）
            if os.path.basename(ws_norm) in _norm_cwd(s) and _norm_cwd(s) != ws_norm:
                off.append("%s → %s" % (a[1][:24], s))
    if off:
        rec("cwd", "fail", "cwds 与本工作区不同形",
            "裂组且会自我强化（去重键＝trim+toLowerCase）：%s" % "；".join(off[:3]))
    else:
        rec("cwd", "ok", "cwds 归属同形", "本线排期的 cwds 与工作区路径逐字一致")

    # ── ④ 活会话：三类当前有没有活的 ─────────────────────────────────────
    live = {}
    for lb, pre in ROLES:
        hits = [s for s in sess if (s[1] or "").startswith(pre) and (s[2] or "") == "working"]
        live[lb] = hits
    dead_roles = [lb for lb, _ in ROLES if not live.get(lb)]
    if dead_roles:
        rec("live", "warn", "当前无活会话",
            "这些类别此刻没有任何活着的会话（按需创建属正常；但主会话开工阶段须建齐）：%s"
            % "、".join(dead_roles))
    else:
        rec("live", "ok", "三类会话均有活的",
            "；".join("%s×%d" % (lb, len(live[lb])) for lb, _ in ROLES))

    # ── ⑤ 投递在跑：协作程序 / 心跳新鲜度 ────────────────────────────────
    hb = os.path.join(WS, ".workbuddy", "collab", "logs", "supervise-heartbeat.json")
    hb_age = None
    for cand in (hb, os.path.join(WS, ".workbuddy", "collab", "supervise-heartbeat.json")):
        try:
            if os.path.isfile(cand):
                hb_age = time.time() - os.path.getmtime(cand)
                break
        except Exception:
            pass
    if hb_age is None:
        rec("deliver", "warn", "投递（常驻）未见心跳",
            "没找到心跳文件 ⇒ 无法证明投递在跑（⛔ 不等于它一定没跑；载体＝专用容器会话）")
    elif hb_age > 900:
        rec("deliver", "fail", "投递心跳陈旧",
            "最后心跳 %.0f 分钟前 ⇒ 常驻可能已掉线" % (hb_age / 60.0))
    else:
        rec("deliver", "ok", "投递心跳新鲜", "最后心跳 %.0f 秒前" % hb_age)

    # ── ⑥ 死排期：`once` 已消耗 **且从未运行** ⇒ 活静默消失 ────────────────
    #    🔴 判据要点两条（都是实测校正出来的，⛔ 别想当然）：
    #      (a) **"已跑完"不是问题、"从未跑却失效"才是**（首版把两者混报 ⇒ 18 条正常痕迹被当 18 个问题）；
    #      (b) **"有没有跑过"必须查 `automation_runs`，⛔ 不能查 `automations.last_run_at`** ——
    #          该字段宿主**根本不写**（实测：本会话自己那条 `接续 · 会话机制合并包 · 任务4b-4d-6`
    #          明明在跑，`last_run_at` 仍是 `None`）⇒ 拿它当判据 = 把"跑过的"全报成"没跑过"。
    try:
        ran = {r[0] for r in c.execute(
            "select distinct automation_id from automation_runs limit 5000")}
    except Exception:
        ran = None
    never_ran, consumed, unverifiable = [], [], []
    for a in mine:
        if a[3] != "once" or a[4]:
            continue
        if ran is None:
            unverifiable.append((a[1] or "")[:30])
        elif a[0] in ran:
            consumed.append((a[1] or "")[:30])
        else:
            never_ran.append((a[1] or "")[:30])
    if never_ran:
        rec("zombie", "fail", "一次性排期从未运行就失效",
            "共 %d 条 ⇒ 这些活会静默消失（消耗掉的一次性排期不会自动重跑）：%s"
            % (len(never_ran), "、".join(never_ran[:4])))
    elif unverifiable:
        rec("zombie", "warn", "无法核对一次性排期是否运行过",
            "读不到 automation_runs（⛔ 不等于它们没跑）；本线有 %d 条 once＋无下次触发：%s"
            % (len(unverifiable), "、".join(unverifiable[:3])))
    elif consumed:
        rec("zombie", "ok", "无「从未运行」的一次性排期",
            "本线 %d 条已跑完的一次性排期（正常痕迹，不计问题）" % len(consumed))
    else:
        rec("zombie", "ok", "无残留的一次性排期", "本线没有「once 且不会再触发」的排期")

    return finish()


def _safe_list(v):
    """cwds 字段形态不稳（JSON 数组 / 逗号串 / 单值）⇒ 一律解成 list[str]。"""
    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 finish():
    bad = [c for c in CHECKS if c["level"] == "fail"]
    warn = [c for c in CHECKS if c["level"] == "warn"]
    verdict = "fail" if bad else ("warn" if warn 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(bad), "warn": len(warn), "ok": len(CHECKS) - len(bad) - len(warn)},
        "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).encode("utf-8"))
        sys.stdout.buffer.write(b"\n")
    elif not QUIET:
        icon = {"ok": "✅", "warn": "⚠️", "fail": "🔴"}[verdict]
        buf = ["%s [会话规划] %s（fail %d / warn %d）" % (icon, summary, len(bad), len(warn))]
        for c in CHECKS:
            m = {"ok": "✓", "warn": "⚠", "fail": "✗"}[c["level"]]
            buf.append("    %s %s —— %s" % (m, c["title"], c["detail"]))
        buf.append("    · 状态标记 → %s" % OUT)
        sys.stdout.buffer.write(("\n".join(buf) + "\n").encode("utf-8"))
    sys.stdout.flush()
    # 退出码语义：**只有"齐备且生效"才 0** —— warn 也要能被机器判出来（⛔ 不用它拦开工，只做标记）
    return 0 if verdict == "ok" else 1


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)