Files
admin 19101acd65 init: workbuddy_skills 重建,仅收录 session-mechanism
- 按用户指示清空原有 25 技能内容,只提交 session-mechanism(57 文件)
- 附 .gitignore(产物 + 本机凭据)
- 令牌明文已脱敏(历史 .neodata_token 与 pitfalls 引用均不入库)
- 本提交为孤儿提交(父提交为空),历史自此重新开始
2026-10-05 14:13:24 +08:00

420 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
lock-guard-hook.py — 「无锁不许改库」的强制钩子(WorkBuddy PreToolUse / SessionStart)
为什么需要它
────────────
2026-09-12 实证:`handoff-guard.sh` 输出「无全局锁」被会话读成「环境干净,可以开工」
(正确读法是「你快去抢锁」),结果**两个会话同时改了本库**。措辞已在 guard 与
`05-交接单/README.md` 里补正,但**约定拦不住不看的人** —— 本脚本是强制层。
作用域(**刻意收窄:对其他项目零影响**)
────────────
仅当 `Write` / `Edit` 的目标路径落在下列根之内,才做锁判定;其余一律放行:
· 文档库 <DSH_DOCS_ROOT>(默认 E:\\ProgramData\\AIProject\\ai1net-dsh-server\\dsh-server-docs)
· 代码库 <DSH_CODE_REPO>(默认 D:\\github\\dsh_shenxian)
⚠️ 例外:路径中含 `.workbuddy` 目录段的一律放行 —— 会话记忆 / 自动化工作数据属“运行态数据”,不是“仓库内容”。(2026-09-13 实测:「代码仓三方同步(DSH)」自动化写自己的 memory/*.md 时被本钩子 deny,属误伤)
判据
────
`<文档库根>/05-交接单/.exec-lock` 存在 = 有人持锁 → 放行(归属由台账 OWNER 承担);
不存在 = **deny**,并把可直接粘贴的抢锁命令回给 Agent。
为什么**不拦 Bash**
──────────────────
① 抢锁命令本身必须能跑(否则把自己锁死 —— 拿不到锁就永远开不了工);
② Bash 写文件是少数派、且破坏面可见(git status);
③ 宁可留一个显式的安全阀,也不要一个可能导致死锁的强制层。
用法(settings.json 的 hooks 段,见同目录 README 或档案 73)
"PreToolUse": [{ "matcher": "Write|Edit", "hooks": [{ "type": "command", "command": "<python> <此脚本>", "timeout": 10 }] }]
"SessionStart":[{ "matcher": "startup", "hooks": [{ "type": "command", "command": "<python> <此脚本>", "timeout": 10 }] }]
退出码:始终 0(判定通过 JSON 输出表达);脚本自身异常也放行,绝不误伤。
"""
# [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()
import json
import os
import sys
import time
# 🔴 2026-10-02:环境定位/体检收敛到 `_env`(同目录优先)——⛔ 不再用 `~` 直拼
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
try:
import _env
except Exception:
sys.stderr.write('[env] 无法导入 _env.py ⇒ 环境定位不可信\n')
raise
# ⚠️ 文档库根:env/roots.env 优先;末位兜底**故意保留字面量** —— 锁守卫一旦静默解析不到,
# 就会"看起来在守、其实没守"(比指错更危险)。换机器请让 install.py 写 roots.env。
DOCS_ROOT = os.environ.get("DSH_DOCS_ROOT") or r"D:\github\dsh_shenxian\dsh-server-docs"
# ⛔ 不留盘符:代码库=文档库的上一级(`<repo>/dsh-server-docs` 的约定)
CODE_REPO = os.environ.get("DSH_CODE_REPO") or (os.path.dirname(DOCS_ROOT) if DOCS_ROOT else "")
LOCK_DIR = os.path.join(DOCS_ROOT, "05-交接单", ".exec-lock")
OWNER_FILE = os.path.join(LOCK_DIR, "OWNER")
LOCKS_ROOT = os.path.join(DOCS_ROOT, "05-交接单", ".locks")
GUARD_CMD = 'bash 07-scripts/handoff-guard.sh --claim-exec "<你的会话名>"'
GUARD_DOMAIN_CMD = 'bash 07-scripts/handoff-guard.sh --claim-exec "<你的会话名>" --domains <域...>'
# ⚠️ 顺序有意义:**先长后短**,与 handoff-guard.sh 的 _ANCHOR_SEGS 逐字一致。
# 两侧只要顺序或集合不同 ⇒ 同一文件算出不同域键 ⇒ 域锁静默失效(假绿)。
# 🔴🔴 2026-10-03 10:0x **修一处真缺陷(不是文档漂移)**:本行原为
# `… "docs", "07-scripts", "08-skills", "05-交接单"`,
# 而真源 `handoff-guard.sh:66 _ANCHOR_SEGS` 是 `… "docs", "scripts", "skills", "交接单"`。
# ⚠️ 后果不是"看起来不一致":`domain_key()` **拿本表参与实际计算** ⇒ 同一个文件
# 在 hook 侧与抢锁侧会算出**不同域键** ⇒ 两边各抢各的 ⇒ **域锁等于没有(假绿)**。
# ✅ 已与 shell 侧逐字对齐(`scripts`/`skills`/`交接单`=**仓内实际目录名**)。
# ⚠️ 改这里**必须**同时改 `handoff-guard.sh` 的 `_ANCHOR_SEGS` 与 `collabd.py` 的 `_DS_ANCHORS`。
# 📌 判据:`collabd.py` 的 `_DS_ANCHORS_OTHER` 保留旧值作对照,自检会报两侧差异。
_DOMAIN_SEGS = ("dsh-server-docs", "ai1net-dsh-server", "aliyun-dsh-server", "src", "poc", "web", "test", "docs", "scripts", "skills", "交接单")
def _owner_text(d: str) -> str:
try:
with open(os.path.join(d, "OWNER"), encoding="utf-8") as f:
return f.read()
except Exception:
return ""
def _my_name(payload: dict) -> str:
"""本会话名 —— hook 进程拿不到 `ME`(那是 bash 侧变量),依次尝试:
① payload/env 的 `DSH_SESSION_NAME`(handoff-guard 抢锁时会写进 OWNER)
② env `ME`(本地手动调用时)
③ OWNER 里**唯一**写着本 session_id 的那把锁(真多会话下 id 唯一,可安全用它反查)
"""
for k in ("DSH_SESSION_NAME",):
v = os.environ.get(k, "")
if v:
return v
v = os.environ.get("ME", "")
if v:
return v
return ""
def _name_owner_of_id(ids: set, exclude: str = "") -> str:
"""用 session_id 反查"我的那把锁"的会话名(仅当它唯一时才返回,避免多锁歧义)。"""
if not ids or not os.path.isdir(LOCKS_ROOT):
return ""
hits = []
for name in os.listdir(LOCKS_ROOT):
if name.startswith("."):
continue
d = os.path.join(LOCKS_ROOT, name)
if not os.path.isdir(d):
continue
txt = _owner_text(d)
if any(i and i in txt for i in ids):
first = (txt.splitlines() or [""])[0].strip()
if first and first != exclude:
hits.append(first)
hits = list(dict.fromkeys(hits))
return hits[0] if len(hits) == 1 else ""
def _is_mine(owner_txt: str, ids: set, cwd: str, my_name: str = "") -> bool:
"""归属判据 —— ⚠️ 会话名与 session_id **同时**吻合才算我的。
2026-09-22 实测踩到:同一进程内用 `ME=会话A/会话B` 模拟两会话时,`CODEBUDDY_SESSION_ID`
是**真实环境变量**、两份 OWNER 都写着同一个 id ⇒ 只按 id 判会把两会话都认作"自己"
⇒ 域隔离**全部放行**(假绿,最危险的失效形态)。
⇒ 判据收紧为:**会话名相等** 且(id 命中 或 未取到 id)。
"""
first = (owner_txt.splitlines() or [""])[0].strip()
id_hit = bool(ids) and any(i and i in owner_txt for i in ids)
if my_name:
return first == my_name and id_hit
if id_hit:
return True
if cwd:
nc = norm(cwd)
if nc and nc in norm(owner_txt):
return True
return False
def _iter_domain_locks():
"""遍历所有域锁目录(跳过 .gate / .migrations / .publish)。"""
if not os.path.isdir(LOCKS_ROOT):
return
for name in os.listdir(LOCKS_ROOT):
if name.startswith("."):
continue
d = os.path.join(LOCKS_ROOT, name)
if not os.path.isdir(d):
continue
try:
with open(os.path.join(d, "DOMAINS"), encoding="utf-8") as f:
doms = [ln.strip() for ln in f if ln.strip()]
except Exception:
continue
yield d, _owner_text(d), doms
def domain_key(path: str) -> str:
"""路径 → 域键(前两段,与 handoff-guard.sh norm_domain 同规则 ⇒ 粒度到目录,宁可保守)。"""
n = norm(path).replace("\\", "/")
parts = [p for p in n.split("/") if p]
for seg in _DOMAIN_SEGS:
if seg in parts:
i = parts.index(seg)
return "/".join(parts[i:i + 2]).lower()
return "/".join(parts[-2:]).lower() if len(parts) >= 2 else n.lower()
def has_my_domain_lock(ids: set, cwd: str, my_name: str = "") -> bool:
mn = my_name or _name_owner_of_id(ids)
for d, ot, _ in _iter_domain_locks():
if _is_mine(ot, ids, cwd, mn):
return True
return False
def blocked_by_domain(path: str, ids: set, cwd: str, my_name: str = "") -> tuple:
"""目标是否落在**别人**的域内 ⇒ (占用者, 冲突域)。自己的域锁一律放行。"""
key = domain_key(path)
mn = my_name or _name_owner_of_id(ids)
for d, ot, doms in _iter_domain_locks():
if _is_mine(ot, ids, cwd, mn):
continue
if key and key in doms:
return ((ot.splitlines()[0] if ot else "?"), key)
return ("", "")
def norm(p: str) -> str:
try:
return os.path.normcase(os.path.normpath(p))
except Exception:
return p
PROTECTED = [norm(DOCS_ROOT), norm(CODE_REPO)]
def is_protected(path: str) -> bool:
if not path:
return False
n = norm(path)
# 会话记忆 / 自动化工作数据不是「仓库内容」:写它不该被本钩子拦(2026-09-13 新增)
if ".workbuddy" in n.split(os.sep):
return False
for root in PROTECTED:
if n == root or n.startswith(root + os.sep):
return True
return False
def lock_owner() -> str:
try:
with open(OWNER_FILE, encoding="utf-8") as f:
return (f.readline() or "").strip() or "(未写 OWNER)"
except Exception:
return "(未写 OWNER)"
def out(obj: dict) -> None:
# ⚠️ 2026-09-16 修复(**根因**):必须走 **`sys.stdout.buffer` 写 bytes**,⛔ 不能走文本层
# (`sys.stdout.write(str)`)。
# 实证:宿主 spawn 本脚本时 stdout 编码**不保证是 UTF-8**(本机 `PYTHONUTF8=1` 在 hook 环境下
# 不保证生效),而 deny 文案里含 `⛔` / `🔐` 等字符 ⇒ 文本层写抛 `UnicodeEncodeError` ⇒
# 异常冒泡、**stdout 为空** ⇒ 宿主**收不到 deny** ⇒ **静默放行**。
# 而 `hook_log()` 已经把 `PreToolUse-deny` 写进日志 ⇒ 表现成 **「日志里有 DENY,但拦不住」**
# (2026-09-16 受控自测:无锁时 Write 文档库 ⇒ 文件真被创建)。
# 对照:`bash-output-guard.py` 用 `sys.stdout.buffer.write(bytes)` ⇒ **实测拦得住**。
# 📌 教训:本脚本的 `stdin` 早前已按 buffer 加固(见 main 顶部注释),**stdout 漏了** ——
# 「读写要同时加固」,只修一半等于没修。
data = json.dumps(obj, ensure_ascii=False).encode('utf-8')
try:
sys.stdout.buffer.write(data)
sys.stdout.buffer.flush()
except Exception:
try:
sys.stdout.write(data.decode('utf-8', 'replace'))
sys.stdout.flush()
except Exception:
pass
LOG_FILE = os.environ.get(
"DSH_LOCK_HOOK_LOG", os.path.join(os.path.dirname(DOCS_ROOT), ".workbuddy", "lock-hook.log")
)
def hook_log(event: str, detail: str) -> None:
"""低频自证日志:只在 SessionStart 与 deny 时写一行 —— 用来回答「hook 到底有没有被触发」。
为什么需要:hook 配置是**启动时缓存**的,改完必须完全重启才加载;没有日志就只能靠猜。
写入失败一律静默(hook 绝不能因为自己出问题而干扰工作)。
"""
try:
os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')}\t{event}\t{detail}\n")
except Exception:
pass
def main() -> None:
try:
try:
# ⚠️ 必须走 buffer 显式 UTF-8:本机环境设了 PYTHONUTF8=1,但只要有人给脚本加 `-E`
# 就会被屏蔽 ⇒ sys.stdin 回退 cp936 ⇒ 含中文路径(`E:\ProgramData\AIProject\…`)的
# payload 解析即炸。而本函数是 **fail-open**(读不懂就放行)⇒ 会**静默失效**:
# 锁守卫不再拦人,却没有任何痕迹。2026-09-15 实测(同款坑已在 stop-dialog-guard.py 踩过)。
raw = sys.stdin.buffer.read().decode("utf-8", "replace")
except Exception:
raw = sys.stdin.read()
payload = json.loads(raw) if raw.strip() else {}
except Exception as e:
# 不留痕 = 失效无声(判不出"没被调用"与"被静默放行")⇒ 必须记一行
hook_log("payload-unparsable", "%s: %s" % (type(e).__name__, str(e)[:80]))
return # 读不懂 payload → 放行(fail-open 方向正确,但不能无声)
tool = payload.get("tool_name")
event = payload.get("hook_event_name") or payload.get("hook_event") or ""
# ★ 入口即留痕(2026-09-16 加)—— 与 stop-dialog-guard.py / skill-load-guard.py 同款。
# 为什么不只记 SessionStart 与 deny:2026-09-16 实测 —— 重启后本钩子**一行都没写**,
# 当时无法区分"没被宿主调用"与"被静默放行",只能靠猜(最后查明:SessionStart 的
# `matcher: "startup"` 不匹配"恢复会话",钩子压根没被执行)。
# 留痕里带上**实际收到的字段名**(keys):下次宿主字段一变,日志里立刻能看出来,
# 而不是等某天发现守卫早已失效。低频:每次调用一行。
hook_log("entry", "event=%s|tool=%s|keys=%s" % (
event or "?", tool or "-", ",".join(sorted(payload.keys()))[:220]))
# ── SessionStart:只提示(该事件的输出只给用户看,不会进 Agent 上下文)──
if event == "SessionStart" or (tool is None and "source" in payload):
src = payload.get("source", "?")
_sid = payload.get("session_id", "")
_ids = {_sid} if _sid else {x for x in (os.environ.get("CODEBUDDY_SESSION_ID", ""),) if x}
_cwd = payload.get("cwd") or ""
if has_my_domain_lock(_ids, _cwd):
msg = "🔐 你持有域锁 —— 只有落进**别人**域的写入会被拦;锁总览:`handoff-guard.sh --locks`"
hook_log("SessionStart", f"本会话持域锁 source={src}")
elif os.path.isdir(LOCK_DIR):
who = lock_owner()
msg = f"🔐 全局执行锁【已被占用】:{who} —— 但**域锁可绕开它**:不重叠的域仍能并行。"
hook_log("SessionStart", f"旧全局锁被占 owner={who} source={src}")
elif os.path.isdir(DOCKS := os.path.join(DOCS_ROOT, "05-交接单", ".locks")) and os.listdir(DOCKS):
msg = "🔐 已有域锁在跑 —— 开工前先抢你要的域(不重叠即可并行):" + GUARD_DOMAIN_CMD
hook_log("SessionStart", f"有域锁无全局锁 source={src}")
elif os.path.isdir(DOCS_ROOT):
msg = "🔐 无锁【空闲】—— 「空闲」≠「可以开工」:先判定可锁定范围再抢锁 →\n " + GUARD_DOMAIN_CMD
hook_log("SessionStart", f"空闲 source={src}")
else:
msg = ""
if msg:
out({"systemMessage": msg, "suppressOutput": True})
return
# ── PreToolUse:真正的强制点 ──
if tool not in ("Write", "Edit", "MultiEdit", "NotebookEdit"):
return
ti = payload.get("tool_input") or {}
fp = ti.get("file_path") or ti.get("path") or ti.get("notebook_path") or ""
if not is_protected(fp):
return
if not os.path.isdir(DOCS_ROOT):
return # 库不在这台机器上 → 与本约定无关,放行
sid = payload.get("session_id") or ""
cwd = payload.get("cwd") or ""
# ⚠️ **payload 有 session_id ⇒ 只用它**,⛔ 不要与 env 取并集!
# 并集会让"payload 里的伪造/过期 id"叠加进程 env 里的真实 id ⇒ 假会话被判成"我" ⇒ 放行(实测踩过)。
# env 仅作**缺省兜底**(payload 未带 session_id 时)。
ids = {sid} if sid else {x for x in (os.environ.get("CODEBUDDY_SESSION_ID", ""),) if x}
my_name = payload.get("session_name") or _my_name(payload)
# ① 旧全局锁存在 ⇒ 兼容:只对**未持域锁**的会话放行(旧行为),
# 已持域锁的会话仍走 ② 做域隔离(2026-09-22 修正)
# ⚠️ 修正原因:首版让「旧锁存在 ⇒ 一律放行」⇒ 域隔离**完全失效**(实测命中)。
legacy_open = os.path.isdir(LOCK_DIR)
if legacy_open and not has_my_domain_lock(ids, cwd, my_name):
return
# ② 域锁路径(2026-09-22 加):自己持域锁 ⇒ 只看是否落进**别人**的域
if has_my_domain_lock(ids, cwd, my_name):
who, dom = blocked_by_domain(fp, ids, cwd, my_name)
if not who:
return # 在我自己的域内(或无人占该域)⇒ 放行
hook_log("PreToolUse-deny-domain", f"{tool} {fp} ← 域 {dom} 属 {who}")
reason = (
"⛔ 被「域锁」拦下:目标文件落进了**别的会话**的冲突域。\n\n"
f"冲突域:{dom}\n占用会话:{who}\n\n"
"域锁的设计意图 = **不重叠的域可以并行**,重叠的域必须串行。\n"
"你有两个合规选择(⛔ 都不得删对方锁,R9):\n"
" ① 等对方释放该域;\n"
" ② **改做不重叠的域**(这才是域锁的价值所在)。\n\n"
f"查全部锁:bash 07-scripts/handoff-guard.sh --locks"
)
out({
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": reason,
},
"systemMessage": f"⛔ 已拦下跨域写入(域 {dom} 属 {who})",
"suppressOutput": True,
})
return
# ③ 既无全局锁、也无自己的域锁 ⇒ 拒写(与旧行为一致)
hook_log("PreToolUse-deny", f"{tool} {fp}")
reason = (
"⛔ 被「锁机制」拦下:本仓库(文档库 / 代码库)当前**你没有持任何锁**,"
"不允许直接修改。\n\n"
"「无锁」的正确读法 = **「你快去抢」**,不是「可以开工」"
"(2026-09-12 实证:两个会话把「无锁」读成「环境干净」→ 同时改了本库)。\n\n"
"请先执行(Bash,不受本钩子限制)——**两种锁选一种**:\n\n"
" 【域锁】只改局部的模块/文件(推荐,可与他人并行)\n"
f" cd \"{DOCS_ROOT}\"\n {GUARD_DOMAIN_CMD}\n\n"
" 【全局锁】无域可声明,或不确定(旧行为,独占)\n"
f" cd \"{DOCS_ROOT}\"\n {GUARD_CMD}\n\n"
"抢到 = 开工许可;**抢不到 = 冲突域被占 → 停手或改做不重叠的域**。\n"
"开工前先判定可锁定范围:`bash 07-scripts/preflight-lock.sh \"<会话名>\" <目标文件...>`。"
)
out(
{
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": reason,
},
"systemMessage": "⛔ 已拦下一次无锁写入(本仓库要求先抢锁)",
"suppressOutput": True,
}
)
if __name__ == "__main__":
try:
main()
except Exception:
pass # 任何异常都不阻断工作
sys.exit(0)