workbuddy_skills/session-mechanism/scripts/hooks/_env.py

# -*- coding: utf-8 -*-
"""**环境定位与环境自检**（2026-10-02 · 用户定案「技能使用时必须先检查环境配置」）。

一句话：把「技能库根 / 配置目录」从**各脚本各自推导**改成**一处判据＋一处体检**。

## 为什么要有这个模块（2026-10-02 实测事故）

`reply-style-guard.py` 曾被注册成文档库 `07-scripts/reply-style-guard.py`（D 盘旧副本）。
它沿 `__file__` 上溯 4 级找 `skills/` ⇒ 从 D 盘够不到 ⇒ 落进最后一行
`os.path.expanduser('~/.workbuddy/skills')`。
⚠️🔴 **Windows 上 `~` 不等于真配置目录**（本机真值在 E 盘 `CODEBUDDY_CONFIG_DIR`）⇒ 那一行是**错的**。

后果比报错更坏：该钩子的设计判据是「哪一级都读不到就**静默零输出**、不影响任何人」
⇒ **「机制坏了」和「没配规则」表现完全一样**，日志里只有一行 `core=0 字符`。
（真实代价：另一个工作区为这件事绕了两轮，在「规则文件在不在」上打转。）

## 三条硬规矩（本模块的判据）

1. 🔴 **顺序固定**：`DSH_SKILLS_ROOT` env → `CODEBUDDY_CONFIG_DIR` env（+ `roots.env` 兜底）
   → 沿 `__file__` 上溯找含 `agent-operating-rules` 的那一层 → ⛔ **不再回落到 `~`**。
2. 🔴 **找不到就报错，不静默**：写 **stderr** ＋ 返回 `None`；调用方据此**非零退出**。
   ⛔ 绝不「静默零输出」—— 那正是把两类事故压成一个表象的元凶。
3. 🔴 **体检与环境标记分离**：`env_check()` 只**读**（体检）；`write_env_stamp()` 才**写**标记。
   ⇒ 体检永远不该有副作用（否则「查一下」就把标记刷新了，标记就永不报警）。

## 标记文件（用户 2026-10-02 要求：「在对应文件夹创建状态变量和时间作为标记」）

落在**配置作用域对应的文件夹**里，文件名 `env-stamp.json`：

- 全局作用域 ⇒ `<配置目录>/env-stamp.json`（本机 `E:/ProgramData/.workbuddy/env-stamp.json`）
- 工作区作用域 ⇒ `<工作区>/.workbuddy/env-stamp.json`

字段：`scope`（global/workspace）｜`checked_at`（ISO 到秒）｜`ok`（bool）｜
`config_dir` / `skills_root`（**判据读数**，便于事后核）｜`hooks`（逐条注册的存在性）｜
`fixed`（本轮修过的问题清单）
"""

from __future__ import annotations

import json
import os
import sys
import time
from pathlib import Path

_STAMP_NAME = "env-stamp.json"
_ROOTS_ENV = "roots.env"          # 技能根的权威落点（install.py 生成）
# 🔴🔴 2026-10-04 单包自包含改造：识别特征改成**多个候选**，⛔ 不再只认
#    `agent-operating-rules`（那是**另一个**技能 ⇒ 只装本包时会把真技能库判成"不存在"）。
#   🔴 旧写法的病：`skills_root()` / `env_check()` 三处都拿`(root)/_SNIFF).is_dir()` 当
#      "技能库在不在"的判据 ⇒ **单包环境下永远判否**（实测：隔离库只搬本包时全部降级）。
#   ✅ 正解：任一候选目录存在即认；⛔ 全都没有才算定位失败（硬规矩②「找不到就报错不静默」不变）。
#      判据数组**第一个元素 = 本包自己** ⇒ 只拷本包也能定位。
_SNIFFS = ("session-mechanism",)


def _has_skill(root) -> bool:
    """技能库根的判据 =⛔ 至少存在一个识别特征目录（⛔ 只认目录真存在，不猜）。"""
    if not root:
        return False
    try:
        return any((Path(_norm(root)) / s).is_dir() for s in _SNIFFS)
    except Exception:
        return False


# ⛔ 保留旧名给外部引用（⛔ 外部脚本若 import 它做文案，仍能拿到第一个候选）
_SNIFF = _SNIFFS[0]


def _norm(p) -> str:
    s = str(p or "").strip()
    return os.path.normpath(s).replace("\\", "/") if s else ""


def _load_roots_env() -> dict:
    """读技能包里的 `roots.env`（⛔ 各脚本的 `_sm_load_roots()` 注入 env 前的原始来源）。"""
    out = {}
    d = Path(__file__).resolve().parent
    for _ in range(5):
        f = d / _ROOTS_ENV
        if f.is_file():
            try:
                for ln in f.read_text(encoding="utf-8").splitlines():
                    ln = ln.strip()
                    if ln and not ln.startswith("#") and "=" in ln:
                        k, v = ln.split("=", 1)
                        out[k.strip()] = v.strip()
                return out
            except Exception:
                return {}
        n = d.parent
        if n == d:
            break
        d = n
    return out


def _norm_ws(p) -> str:
    """工作区路径**归一化**（判归属专用）。

    🔴 为什么不能直接比字符串：`sessions.cwd` / `automations.cwds` 里
    **同一区出现过反斜杠与正斜杠两种写法**（`E:\\ProgramData\\…` vs `E:/ProgramData/…`），
    末尾斜杠有时在有时不在 ⇒ **逐字比会把同一个区判成两个** ⇒ 归属误判、裂组。
    ⇒ 统一：转正斜杠、去尾斜杠、转小写。
    """
    s = str(p or "").strip().replace("\\", "/")
    while s.endswith("/") and len(s) > 3:
        s = s[:-1]
    return s.lower()


def config_dir() -> str:
    """**真配置目录** —— 顺序见硬规矩①的前三档。"""
    v = os.environ.get("CODEBUDDY_CONFIG_DIR")
    if v:
        return _norm(v)
    v = _load_roots_env().get("CODEBUDDY_CONFIG_DIR")
    if v:
        return _norm(v)
    #⛔ 最后一档仍不给 `~`：Windows 上它通常不是真配置目录（实测会指到 C:\Users\...）
    return _norm(Path.home() / ".workbuddy")


def skills_root() -> str:
    """**技能库根** —— 顺序见硬规矩①。⛔ 不再回落到 `~`（硬规矩①末档已删）。

    🔴 2026-10-02 补：**每一档都要验"真的存在"**，⛔ 不信任 env 值本身。
       实测踩过：`DSH_SKILLS_ROOT` 指向一个不存在的目录时，旧写法直接返回它
       ⇒ 下游 `os.path.isfile(核心块)` 失败 ⇒ **静默零输出**（同款静默失效）。
       ⇒ env 给错 ⇒ 继续往下找；全都找不到 ⇒ 返回 `""`（＝报错，⛔ 不猜）。
    """
    v = os.environ.get("DSH_SKILLS_ROOT")
    if v and _has_skill(v):
        return _norm(v)
    d = Path(__file__).resolve().parent
    for _ in range(6):                       # 沿 __file__ 上溯（换机可用，⛔ 不写死盘符）
        cand = d.parent
        if _has_skill(cand):
            return _norm(cand)
        if d.parent == d:
            break
        d = d.parent
    r = _load_roots_env().get("CODEBUDDY_CONFIG_DIR")
    if r:
        cand = Path(_norm(r)) / "skills"
        if _has_skill(cand):
            return _norm(cand)
    return ""                                  # 🔴 空串＝找不到（⛔ 不是"猜一个"）


def _stamp_path(ws: str = "", scope: str = "global") -> str:
    if scope == "workspace" and ws:
        return _norm(Path(ws) / ".workbuddy" / _STAMP_NAME)
    return _norm(Path(config_dir()) / _STAMP_NAME)


def registered_hooks() -> list:
    """读全局 `settings.json` 里已注册的钩子 ⇒ `[(事件, 命令, timeout, 脚本路径)]`。"""
    sp = Path(config_dir()) / "settings.json"
    out = []
    try:
        d = json.loads(sp.read_text(encoding="utf-8"))
    except Exception:
        return out
    for ev, groups in (d.get("hooks") or {}).items():
        for g in (groups or []):
            for e in (g or {}).get("hooks", []):
                c = (e or {}).get("command", "") or ""
                # 取最后一个 .py 片段（命令形如`<python>"<脚本>" <参数>`）
                parts = [p for p in c.split('"') if p.strip().endswith(".py")]
                out.append((ev, c, (e or {}).get("timeout"),
                            _norm(parts[-1]) if parts else ""))
    return out


def env_check(ws: str = "", scope: str = "global") -> dict:
    """🔍 **只读体检**：返回逐项读数，⛔ **绝不写标记**（硬规矩③）。

    🔴🔴 2026-10-04 补上**工作区部署**那一半（用户：「使用技能时就要检查环境、做相关配置，
       ⚠️ **说了无数遍但没实现**」）。**为什么必须补**：
       原来的 `env_check` 只查两件事 —— 技能根能不能定位、钩子有没有注册
       ⇒ **「技能被复制进某个工作区、但那个工作区根本没部署」这类状态它永远查不出来**
       ⇒ 没人提醒、没人负责 ⇒ 机制在该工作区**跑不起来却看不出为什么**。
       🔴 实测坐实：`vibe-product` 复制了完整副本（47 个文件），但**无 `goal.json`、
       无工作区级 `collabd.py`、配置是旧格式** ⇒ 旧版 `env_check` 读数全绿。
    """
    rep = {"scope": scope, "checked_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
           "config_dir": config_dir(), "skills_root": skills_root(), "bad": [],
           "warn": []}
    if not rep["skills_root"]:
        rep["bad"].append("skills_root 定位失败（⛔ 已不回落~）")
    elif not _has_skill(rep["skills_root"]):
        rep["bad"].append("skills_root 存在但缺识别特征目录 %s：%s"
                           % ("/".join(_SNIFFS), rep["skills_root"]))
    hooks = []
    for ev, cmd, t, path in registered_hooks():
        ok = bool(path) and os.path.isfile(path)
        row = {"event": ev, "timeout": t, "path": path, "exists": ok}
        hooks.append(row)
        if not ok:
            rep["bad"].append("钩子目标不存在：%s %s" % (ev, path or cmd[:60]))
    rep["hooks"] = hooks
    # ── 🔴🔴 补：**工作区部署**体检（原来完全没有这一段）──────────────────
    # ⚠️ `ws` 传空就**回落成技能根的父目录**（⛔ 不猜：副本场景下技能根在
    #    `<工作区>/.workbuddy/skills/<包>/`，往上一级才是工作区）。
    _ws = _norm(ws or "")
    if not _ws and rep["skills_root"]:
        _ws = _norm(str(Path(rep["skills_root"]).parent.parent))
    dep = {"workspace": _ws, "deployed": None, "missing": [], "stale": []}
    if not _ws:
        dep["missing"].append("**工作区根定位不到**（⛔ 传 `ws=` 或让技能根可定位）")
    else:
        w = Path(_ws)
        cfg_p = w / ".workbuddy" / "collab" / "collabd.config.json"
        goal_p = w / "tmp" / "supervise-inbox" / "goal.json"
        # ① 部署配置：⛔ 不存在 ＝ 没部署（`collabd.py` 不认这个工作区）
        dep["config"] = {"path": _norm(str(cfg_p)), "exists": cfg_p.exists()}
        if not cfg_p.exists():
            dep["missing"].append("**部署配置不存在**（`collabd.py` 不认这个工作区）"
                                  " ⇒ 跑 `init_workspace.py <工作区>`")
        # ② 目标文件：⛔ 没有它，机制没有目标可推
        dep["goal"] = {"path": _norm(str(goal_p)), "exists": goal_p.exists()}
        if not goal_p.exists():
            dep["missing"].append("**目标文件不存在**（机制没有目标可推）"
                                  " ⇒ 建 `tmp/supervise-inbox/goal.json`")
        # ③ inbox 骨架
        if not (w / "tmp" / "supervise-inbox").is_dir():
            dep["missing"].append("**inbox 骨架缺失**（`tmp/supervise-inbox/`）")
        # ④ 配置**格式是否过期**（有配置却缺现行键 ＝ 旧格式 ＝ 行为不可预期）
        if cfg_p.exists():
            try:
                _c = json.loads(cfg_p.read_text(encoding="utf-8"))
                need = ("log", "supervise_interval", "wake_enable")
                gone = [k for k in need if k not in _c]
                if gone:
                    dep["stale"].append("配置缺现行键 %s ⇒ **旧格式**（升级技能后不会自动补）"
                                        % "/".join(gone))
            except Exception as e:
                dep["missing"].append("部署配置**读不出/解析失败**：%r" % (e,))
    dep["deployed"] = not dep["missing"]
    rep["deploy"] = dep
    # ⚠️ 归类：**没部署进 `warn` 而不塞 `bad`** ——
    #   「这个工作区还没启用机制」是**合法状态**（技能可只作资产存放），
    #   ⛔ 不该让只装了技能副本的工作区一律体检红。查 `deploy.deployed` 即可。
    rep["warn"].extend(dep["missing"] + dep["stale"])

    # 🔴🔴🔴 ⑤ **工作区归属核对**（S 红线 · 2026-10-04 用户当场纠正后补）
    #    病根：用户在本工作区下达「监控 vibe-product」的目標，我把任务会话建到了
    #    `vibe-product` ⇒ 那条会话在本工作区看板/台账里**直接不可见**，
    #    而本工作区恰恰是问题最集中、最需要它的地方。⛔ 没有任何机制提醒过。
    #    ⇒ 这里做成**体检项**：技能一被用起来就核对一次归属，⛔ 不靠人记住。
    att = {"this_ws": _ws, "declared": [], "mismatch": [], "peer_missing": []}
    try:
        _db = Path(config_dir()) / "workbuddy.db"
        if _db.exists():
            import sqlite3
            _con = sqlite3.connect("file:%s?mode=ro" % _db.as_posix(), uri=True, timeout=8)
            try:
                _con.execute("PRAGMA busy_timeout=8000")
                # 读法：⛔ **不按 `sessions.cwd` 分组**（后台会话可带独立 cwd），
                # 而按 `automations.cwds` 的字面正斜杠形态 —— 那才是「这条线归谁」。
                _rows = _con.execute(
                    "SELECT name, cwds, status FROM automations").fetchall()
            finally:
                _con.close()
            _this = _norm_ws(_ws)
            for _nm, _cw, _stt in _rows:
                try:
                    _c = json.loads(_cw or "[]")
                except Exception:
                    _c = []
                for _one in _c:
                    att["declared"].append({"name": _nm, "cwd": _one,
                                            "status": _stt, "is_self": _norm_ws(_one) == _this})
            # 🔴 判「有没有本工作区自己的线」：⛔ **只数有名字的**（按库实测：
            #    机制里存在多条**无 `name` 的排期** ⇒ 拿来凑数会恒绿）
            att["self_lines"] = [d for d in att["declared"] if d["is_self"]]
        else:
            att["peer_missing"].append("宿主库 `%s` 读不到（⛔ 归属无法核对）" % _db)
    except Exception as e:
        att["peer_missing"].append("归属核对失败：%r" % (e,))
    if not att.get("self_lines") and not att["peer_missing"]:
        att["peer_missing"].append(
            "**本工作区名下没有任何协作线**（`automations.cwds` 里找不到它）"
            " ⇒ 目标/会话会落到别的工作区去，本区看板与台账看不到")
    rep["attribution"] = att
    rep["warn"].extend(att["peer_missing"])

    rep["ok"] = not rep["bad"]
    return rep



def write_env_stamp(ws: str = "", scope: str = "global", fixed=None) -> str:
    """✍️ **写环境标记**（含状态变量＋时间）。`scope` 决定落在哪个文件夹。"""
    rep = env_check(ws, scope)
    rep["fixed"] = list(fixed or [])
    p = Path(_stamp_path(ws, scope))
    try:
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps(rep, ensure_ascii=False, indent=2) + "\n",
                     encoding="utf-8", newline="\n")
    except Exception as e:
        sys.stderr.write("env-stamp 写失败：%r\n" % (e,))
        return ""
    return _norm(p)


def read_env_stamp(ws: str = "", scope: str = "global") -> dict:
    try:
        return json.loads(Path(_stamp_path(ws, scope)).read_text(encoding="utf-8"))
    except Exception:
        return {}


def require_env(ws: str = "", what: str = "") -> str:
    """🔴 **调用方统一入口**：定位不到就**报错并返回空串**（硬规矩②）。

    ⛔ 绝不在这里 `sys.exit` —— 钩子进程退出码由调用方决定（各钩子预算不同）。
    """
    root = skills_root()
    if root:
        return root
    sys.stderr.write(
        "[env] 技能库根定位失败%s：DSH_SKILLS_ROOT / CODEBUDDY_CONFIG_DIR 均未给，"
        "沿 __file__ 上溯也没找到含 %s 的目录。\n"
    "     ⛔ 已⛔静默跳过（不再回落 ~，因Windows 上 ~ 不是真配置目录）\n"
    "     ✅ 修法：设环境变量 DSH_SKILLS_ROOT，或跑 "
    "`python scripts/hooks/_env.py --stamp` 重建标记并体检\n"
        % (("（%s）" % what) if what else "", "/".join(_SNIFFS)))
    return ""


if __name__ == "__main__":
    import argparse
    _ap = argparse.ArgumentParser(description="环境自检／环境标记")
    _ap.add_argument("--ws", default="", help="工作区绝对路径")
    _ap.add_argument("--scope", default="global", choices=["global", "workspace"])
    _ap.add_argument("--stamp", action="store_true", help="写环境标记")
    _ap.add_argument("--fixed", default="", help="--stamp 时记进 fixed 的说明")
    _a = _ap.parse_args()
    if _a.stamp:
        _p = write_env_stamp(_a.ws, _a.scope,
                             [_a.fixed] if _a.fixed else None)
        print("已写标记：%s" % (_p or "⛔ 写失败"))
        raise SystemExit(0 if _p else 1)
    _r = env_check(_a.ws, _a.scope)
    print("配置目录 = %s" % _r["config_dir"])
    print("技能库根 = %s" % (_r["skills_root"] or "⛔ 定位失败"))
    print("已注册钩子 = %d 条｜不存在的 = %d"
          % (len(_r["hooks"]), sum(1 for h in _r["hooks"] if not h["exists"])))
    for _b in _r["bad"]:
        print("  ⛔ %s" % _b)
    print("结论 = %s" % ("✅ 通过" if _r["ok"] else "❌ 不通过"))
    raise SystemExit(0 if _r["ok"] else 1)