357 lines
17 KiB
Python
357 lines
17 KiB
Python
# -*- 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)
|