dsh_shenxian/dsh-server-docs/scripts/lock-guard-hook.py

#!/usr/bin/env python3
"""
lock-guard-hook.py — 「无锁不许改库」的强制钩子（WorkBuddy PreToolUse / SessionStart）

为什么需要它
────────────
2026-09-12 实证：`handoff-guard.sh` 输出「无全局锁」被会话读成「环境干净，可以开工」
（正确读法是「你快去抢锁」），结果**两个会话同时改了本库**。措辞已在 guard 与
`交接单/README.md` 里补正，但**约定拦不住不看的人** —— 本脚本是强制层。

作用域（**刻意收窄：对其他项目零影响**）
────────────
仅当 `Write` / `Edit` 的目标路径落在下列根之内，才做锁判定；其余一律放行：
  · 文档库  <DSH_DOCS_ROOT>（默认 E:\\ProgramData\\AI技能\\aliyun-dsh-server\\dsh-server-docs）
  · 代码库  <DSH_CODE_REPO>（默认 D:\\github\\dsh_shenxian）
  ⚠️ 例外：路径中含 `.workbuddy` 目录段的一律放行 —— 会话记忆 / 自动化工作数据属“运行态数据”，不是“仓库内容”。（2026-09-13 实测：「代码仓三方同步（DSH）」自动化写自己的 memory/*.md 时被本钩子 deny，属误伤）

判据
────
`<文档库根>/交接单/.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 输出表达）；脚本自身异常也放行，绝不误伤。
"""

import json
import os
import sys
import time

DOCS_ROOT = os.environ.get("DSH_DOCS_ROOT", r"D:\github\dsh_shenxian\dsh-server-docs")
CODE_REPO = os.environ.get("DSH_CODE_REPO", r"D:\github\dsh_shenxian")
LOCK_DIR = os.path.join(DOCS_ROOT, "交接单", ".exec-lock")
OWNER_FILE = os.path.join(LOCK_DIR, "OWNER")

GUARD_CMD = 'bash scripts/handoff-guard.sh --claim-exec "<你的会话名>"'


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:
    sys.stdout.write(json.dumps(obj, ensure_ascii=False))
    sys.stdout.flush()


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\AI技能\…`）的
            # 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 ""

    # ── SessionStart：只提示（该事件的输出只给用户看，不会进 Agent 上下文）──
    if event == "SessionStart" or (tool is None and "source" in payload):
        src = payload.get("source", "?")
        if os.path.isdir(LOCK_DIR):
            who = lock_owner()
            msg = f"🔐 全局执行锁【已被占用】：{who} —— 同一时刻只允许一个执行会话动「文档/代码/服务器」。"
            hook_log("SessionStart", f"已被占用 owner={who} source={src}")
        elif os.path.isdir(DOCS_ROOT):
            msg = "🔐 全局执行锁【空闲】—— 但「空闲」≠「可以开工」：动手前先抢锁 → " + GUARD_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 os.path.isdir(LOCK_DIR):
        return  # 有人持锁 → 放行（归属由 OWNER / 台账承担）
    if not os.path.isdir(DOCS_ROOT):
        return  # 库不在这台机器上 → 与本约定无关，放行

    hook_log("PreToolUse-deny", f"{tool} {fp}")

    reason = (
        "⛔ 被「锁机制」拦下：本仓库（文档库 / 代码库）当前**无人持全局执行锁**，"
        "不允许直接修改。\n\n"
        "「无锁」的正确读法 = **「你快去抢」**，不是「可以开工」"
        "（2026-09-12 实证：两个会话把「无锁」读成「环境干净」→ 同时改了本库）。\n\n"
        f"请先执行（Bash，不受本钩子限制）：\n    cd \"{DOCS_ROOT}\"\n    {GUARD_CMD}\n\n"
        "抢到 = 开工许可；**抢不到 = 有会话在跑 → 停手**（输出会告诉你占用者与在做哪单）。\n"
        "若确实只需追加一行、且已确认无人在动，可由用户裁定后临时移除本钩子。"
    )
    out(
        {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": reason,
            },
            "systemMessage": "⛔ 已拦下一次无锁写入（本仓库要求先抢锁）",
            "suppressOutput": True,
        }
    )


if __name__ == "__main__":
    try:
        main()
    except Exception:
        pass  # 任何异常都不阻断工作
    sys.exit(0)