fix(gate): 会话实测二批修复——并发写原子化 + 指纹截断盲区 + UPS 问句句级抑制

基于 zcode 侧 codex-java-sdk 三分支发布全程实战（双副本挂载 + 并行会话共用
工作树场景），codegraph 全量索引后定位的四项不足：

1. dedup/gate-cache 写入原子化：hook_dedup.json 与 gate cache 此前为裸
   read-modify-write——UPS 与 PreToolUse 数秒窗口内先后执行会互相覆盖
   （丢抑制事件→双份报告）或读到半截 JSON（ValueError 静默吞→去重失效）。
   加 fcntl 进程锁（Windows 降级可选）+ tmp+os.replace 原子替换；
   dedup 时间戳从 monotonic 改墙钟（monotonic 跨重启回绕，不能持久化）。
2. _worktree_fingerprint 去掉 [:500] 截断：截断使第 501+ 文件的修复在
   60s 缓存窗口内被旧指纹掩盖（retry-timing 陷阱对大改动面复活）。
   名字集合全量入 hash，stat 叠加不再封顶（每次仅对变更文件 stat，成本可控）。
3. UPS 问句抑制改句级：`这个方案 OK 吗？帮我提交` 问句与祈使混排时整条被
   静默——改为按句子切分，仅当全部触发句均为问句才静默。
4. languages.json 移除 .zsh 扩展（gate_lib 剥离兜住 run_gate 路径，但
   fix.py/run_check 直调仍会送检）；docs/LANGUAGES.md 同步重生成。
5. tests/run_all fail-open 测试免疫本机 codeguard.skipGate 配置——
   此前依赖仓库本地 git config 未设 skipGate，环境污染即假失败（实测复现）。

测试：141 通过 / 0 失败（run_all 全量）；ruff hooks/+tests/ 全过。

Author: loong10k

docs/LANGUAGES.md

@@ -20,7 +20,7 @@
 | PHP | `.php` | `php -l {file}` | `php-cs-fixer fix` | composer require --dev php-cs-fixer |
 | Ruby | `.rb` | `rubocop` | `rubocop -A` | gem install rubocop |
 | Scala | `.scala` `.sc` | `scalafmt --check .` | `scalafmt` | coursier install scalafmt |
-| Shell | `.sh` `.bash` `.zsh` | `shellcheck --severity=warning {file}` | `shfmt -w .` | brew install shellcheck shfmt |
+| Shell | `.sh` `.bash` | `shellcheck --severity=warning {file}` | `shfmt -w .` | brew install shellcheck shfmt |
 | Dockerfile | （文件名匹配） | `hadolint {file}` | `hadolint` | brew install hadolint |
 | YAML | `.yml` `.yaml` | `yamllint .` | `yamllint .` | pip install yamllint |
 | Elixir | `.ex` `.exs` | `mix credo --strict` | `mix format` | 项目需配置 credo 依赖 |

hooks/gate_lib.py

@@ -18,6 +18,11 @@
 import time
 from pathlib import Path
 
+try:  # Windows 无 fcntl——锁降级为可选，原子替换仍生效
+    import fcntl
+except ImportError:  # pragma: no cover
+    fcntl = None
+
 PLUGIN_ROOT = Path(__file__).resolve().parents[1]   # hooks/ 的上级 = 插件根
 sys.path.insert(0, str(PLUGIN_ROOT / "scripts"))
 
@@ -155,18 +160,23 @@ def _worktree_fingerprint(project_root: Path) -> str:
         h.update(("\n".join(names)).encode("utf-8", "replace"))
         if args[:2] == ("diff", "--name-only"):
             # 未暂存已跟踪文件：内容改动名字不变，必须叠加 stat 指纹
-            # （staged 路不需要——暂存内容变化必动 index，由键内 idx_sig 兜底）
-            for name in names[:500]:
+            # （staged 路不需要——暂存内容变化必动 index，由键内 idx_sig 兜底）。
+            # 名字集合不截断：截断会让第 N+1 个文件的修复在缓存窗口内被旧
+            # 指纹掩盖（retry-timing 陷阱对大改动面复活）。stat 本身足够便宜。
+            for name in names:
                 try:
                     st = (project_root / name).stat()
                     h.update(f"{name}:{st.st_size}:{st.st_mtime_ns}".encode("utf-8", "replace"))
                 except OSError:
                     h.update(name.encode("utf-8", "replace"))
     # 未跟踪文件：ls-files 只给文件名——内容改动名字不变，必须叠加 stat 指纹
     others = (_git(project_root, "ls-files", "--others") or "").splitlines()
-    for name in sorted(others)[:500]:
-        if not name.strip():
-            continue
+    for name in sorted(n for n in others if n.strip()):
+        try:
+            st = (project_root / name).stat()
+            h.update(f"{name}:{st.st_size}:{st.st_mtime_ns}".encode("utf-8", "replace"))
+        except OSError:
+            h.update(name.encode("utf-8", "replace"))
         try:
             st = (project_root / name).stat()
             h.update(f"{name}:{st.st_size}:{st.st_mtime_ns}".encode("utf-8", "replace"))
@@ -288,9 +298,13 @@ def run_gate(
 
     if ck:
         with contextlib.suppress(OSError):
-            cache_file.write_text(json.dumps(
+            # 原子替换：UPS 与 PreToolUse 并发写同一 cache 文件时，
+            # 裸 write_text 可能让对方读到半截 JSON（ValueError → 缓存失效重扫）。
+            tmp = cache_file.with_name(f"{cache_file.name}.{os.getpid()}.tmp")
+            tmp.write_text(json.dumps(
                 {"key": ck, "ts": _time.time(), "failures": failures, "skipped": skipped},
                 ensure_ascii=False))
+            os.replace(tmp, cache_file)
     return failures, skipped
 
 
@@ -769,26 +783,33 @@ def should_suppress_event(key: str, window: float = 2.0) -> bool:
     import hashlib
     import time as _time
     path = codeguard_home() / "hook_dedup.json"
-    now = _time.monotonic()
+    now = _time.time()
     try:
         path.parent.mkdir(parents=True, exist_ok=True)
-        dedup = {}
-        if path.exists():
-            dedup = json.loads(path.read_text(encoding="utf-8"))
-        h = hashlib.sha1(key.encode("utf-8", "replace")).hexdigest()[:24]
-        last = dedup.get(h) or {}
-        stale = [k for k, v in dedup.items() if now - v.get("ts", 0) > 60]
-        for k in stale:
-            dedup.pop(k, None)
-        if now - last.get("ts", 0) < window:
-            dedup[h] = last
-            path.write_text(json.dumps(dedup), encoding="utf-8")
-            return True
-        dedup[h] = {"ts": now}
-        path.write_text(json.dumps(dedup), encoding="utf-8")
+        # 并发安全：UPS 与 PreToolUse 两条钩子在数秒窗口内先后读写本文件，
+        # 裸 read-modify-write 会互相覆盖（丢抑制事件 → 双份报告）或读到半截
+        # JSON（ValueError → 去重静默失效）。进程锁 + tmp+replace 原子替换。
+        # 时间戳用墙钟：monotonic 跨重启回绕，持久化文件不能存。
+        lock_path = path.with_name(path.name + ".lock")
+        with open(lock_path, "w") as lock:
+            if fcntl:
+                fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
+            dedup = {}
+            if path.exists():
+                dedup = json.loads(path.read_text(encoding="utf-8"))
+            h = hashlib.sha1(key.encode("utf-8", "replace")).hexdigest()[:24]
+            last = dedup.get(h) or {}
+            stale = [k for k, v in dedup.items() if now - v.get("ts", 0) > 60]
+            for k in stale:
+                dedup.pop(k, None)
+            suppress = now - last.get("ts", 0) < window
+            dedup[h] = {"ts": last.get("ts", now)} if suppress else {"ts": now}
+            tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
+            tmp.write_text(json.dumps(dedup), encoding="utf-8")
+            os.replace(tmp, path)
+        return suppress
     except (OSError, ValueError):
         return False
-    return False
 
 
 def summarize_failures(failures: list) -> str:

hooks/user_prompt_validator.py

@@ -86,13 +86,22 @@ def is_trigger(user_text: str) -> bool:
     """
     if not user_text:
         return False
-    if any(q in user_text for q in QUESTION_MARKERS):
+    # 问句抑制按**句子粒度**：`这个方案 OK 吗？帮我提交` 问句与祈使混排时，
+    # 整条文本级的问号判断会把祈使句一起静默（软门缺失，硬门兜底，但漏提醒
+    # 不符合"软门多提醒不算错"的既定原则）。
+    sentences = [seg for seg in _re.split(r"[。！!？?\n\r]+", user_text) if seg.strip()]
+    trigger_sentences = [seg for seg in sentences if _TRIGGER_RE.search(seg)]
+    if not trigger_sentences:
         return False
-    m = _TRIGGER_RE.search(user_text)
+    non_question = [seg for seg in trigger_sentences
+                    if not any(q in seg for q in QUESTION_MARKERS)]
+    if not non_question:
+        return False
+    m = _TRIGGER_RE.search(non_question[0])
     if not m:
         return False
     # 句首祈使：触发词本身（或前面只有空白/常见介词）位于句首。
-    stripped = user_text.lstrip()
+    stripped = non_question[0].lstrip()
     leading = stripped[: m.start()].strip().lower()
     if leading in ("", "git", "to", "the", "a", "an", "帮我", "请", "请帮我"):
         return True

scripts/languages.json

@@ -369,8 +369,7 @@
       "name": "Shell",
       "extensions": [
         ".sh",
-        ".bash",
-        ".zsh"
+        ".bash"
       ],
       "markers": [],
       "status": "stable",

tests/run_all.py

@@ -576,6 +576,7 @@ def counting(root, cfg, langs, **kw):
         "import gate_lib\n"
         "def _boom(*a, **k): raise RuntimeError('injected-failure')\n"
         "gate_lib.run_gate=_boom\n"
+        "gate_lib.skip_gate_via_git_config=lambda *a, **k: False\n"  # 免疫本机 skipGate 配置
         f"runpy.run_path({str(HOOKS / 'user_prompt_validator.py')!r}, run_name='__main__')\n"
     )
     r = subprocess.run([sys.executable, "-c", code], capture_output=True, check=False, text=True,