Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 77 additions & 12 deletions hooks/gate_lib.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,12 @@
r"Could not resolve dependencies|Could not find artifact|DependencyResolutionException"
)

# .zsh 送检剥离的统一话术:剥离不是静默丢弃——skipped 必须带上可执行的修复
# 指令(ShellCheck 不支持 zsh 方言;文件头加方言声明注释后即可正常送检)。
# delta/全量/UPS 三个面共用这一份认知(改这里即三处同变)。
_ZSH_SKIP_NOTE = ("ShellCheck 不支持 zsh 方言,{n} 个 zsh 文件未送检。"
"修复:在每个 .zsh 文件头添加 `# shellcheck shell=bash` 注释后重试")

from detect_lang import (
LANG_COMMANDS,
detect_language,
Expand Down Expand Up @@ -317,11 +323,32 @@ def _log_path(project_root: Path, lang: str) -> Path:
return Path(tempfile.gettempdir()) / f"codeguard-gate-{key}-{lang}.log"


def _rule_summary(full: str) -> str:
"""ruff/checkstyle 输出的规则聚合计数行(如 "EXE001×3 UP009×1")。

2026-09-23 opencli 会话实测:门禁输出被截断后 AI 修一个才暴露下一个
(whack-a-mole 升级版),聚合计数让 AI 一次看到问题全集与规模。
非 linter 标准行格式(无规则码锚)时返回空串。
"""
import re as _re
counts: dict[str, int] = {}
for line in full.splitlines():
m = _re.search(r"\b([A-Z]{2,5}\d{3,4})\b", line)
if m and (":" in line or "-->" in line):
counts[m.group(1)] = counts.get(m.group(1), 0) + 1
if not counts:
return ""
return "规则汇总: " + " ".join(f"{k}×{v}" for k, v in sorted(counts.items()))


def _truncate_detail(full: str, project_root: Path, lang: str) -> str:
"""节选 + 总量 + 完整日志路径:截断只留头 600 字符会让 AI 一次修 3 个、
重试再看 3 个(whack-a-mole);必须给出"共几行、完整在哪"。"""
lines = [ln for ln in full.splitlines() if ln.strip()]
summary_line = _rule_summary(full)
detail = full[:600]
if summary_line:
detail = summary_line + "\n" + detail
if len(lines) > 8 or len(full) > 600:
try:
path = _log_path(project_root, lang)
Expand Down Expand Up @@ -403,6 +430,9 @@ def _run_gate_uncached(
"""
from concurrent.futures import ThreadPoolExecutor

# 部分剥离场景的 skipped 备注(闭包列表:worker 线程 append,GIL 下安全)
zsh_notes: list[str] = []

def check(lang: str):
cmd_def = LANG_COMMANDS.get(lang)
if not cmd_def:
Expand All @@ -415,7 +445,9 @@ def check(lang: str):
if outcome["status"] == "PASS":
return None
if outcome["status"] != "FAIL":
return (lang, None, f"java {outcome['status']}: {outcome['reason']}")
hint_extra = ";若 wrapper 缺执行位:chmod +x mvnw" \
if "不可执行" in outcome.get("reason", "") else ""
return (lang, None, f"java {outcome['status']}: {outcome['reason']}{hint_extra}")
detail = _truncate_detail(outcome.get("stdout_tail", "") + outcome.get("stderr_tail", ""), project_root, lang)
if outcome.get("log_path"):
detail += f"\n完整输出: {outcome['log_path']}"
Expand All @@ -428,14 +460,17 @@ def check(lang: str):
if detect_language(f, project_root) == lang and (project_root / f).is_file()
]
# ShellCheck 不支持 zsh(SC1071 是 error 级固有限制)——.zsh 送检
# 必红且不是代码违规。从目标面剔除并明示"未验证",不静默丢弃。
# 必红且不是代码违规。从目标面剔除并明示"未验证"+ 可执行修复指令,
# 不静默丢弃。部分 zsh 场景备注进 zsh_skips,余下 .sh 照常送检
# (会话实测:静默剥离让提交方不知道有一部分文件压根没被检查)。
if lang == "shell":
zsh_files = [f for f in lang_files if f.endswith(".zsh")]
if zsh_files:
lang_files = [f for f in lang_files if not f.endswith(".zsh")]
note = _ZSH_SKIP_NOTE.format(n=len(zsh_files))
if not lang_files:
return (lang, None,
f"shell {len(zsh_files)} 个 zsh 文件未验证(ShellCheck 不支持 zsh)")
return (lang, None, note)
zsh_notes.append(note)
if not lang_files:
return (lang, None, f"{lang} 本次改动未涉及,跳过")
uses_delta_files = bool(lang_files) and len(lang_files) <= 50
Expand Down Expand Up @@ -470,7 +505,7 @@ def check(lang: str):
return (lang, None,
(f"{lang} 项目分析 UNVERIFIED({_jp.get('build_system', '?')}),"
f"原因: {'; '.join(_jp.get('reasons', []))},本次未验证"))
exe = _jp.get("executable")
exe = _java_executable_from_plan(_jp)
if exe and base_cmd and base_cmd[0] in ("mvn", "gradle"):
base_cmd = [exe] + base_cmd[1:]
except Exception as exc: # noqa: BLE001 — 分析失败不阻塞门禁,走原路径
Expand Down Expand Up @@ -595,7 +630,7 @@ def _run_one(cmd: list[str], timeout: int, lang: str, hint: str) -> tuple[int, s
skipped.append(openspec_check["skipped"])
except Exception as exc: # noqa: BLE001 — 调用面异常同样归"未验证",不静默吞
skipped.append(f"openspec UNVERIFIED:{exc!r}")
return failures, skipped
return failures, skipped + zsh_notes


def _openspec_validate(project_root: Path, timeout_seconds: int = 300) -> dict:
Expand Down Expand Up @@ -627,6 +662,26 @@ def _openspec_validate(project_root: Path, timeout_seconds: int = 300) -> dict:
hint = "见 https://openspec.dev 或 `npm i -g @fission-ai/openspec`"
return {"failures": [("openspec", detail, fix, hint)], "skipped": None}

def _java_executable_from_plan(plan: dict) -> str | None:
"""从 java_project.analyze 的产物里解析构建可执行文件。

两条来源按优先级:
1. 顶层 ``executable`` 键(java_project 正常产出);
2. 回退 ``commands[0].argv[0]``——历史版本的 java_project 只把
wrapper 体现在计划命令里(实测 68c8a37 曾因此让 mvnw 感知成为
死代码:消费者读的键生产者从不写)。
非 mvn/gradle wrapper(如 codeguard.json 显式命令)不参与替换。
"""
exe = plan.get("executable")
if exe:
return exe
for cmd in plan.get("commands", []) or []:
argv = cmd.get("argv") or []
if argv and Path(argv[0]).name in ("mvnw", "gradlew"):
return argv[0]
return None


def skip_gate_via_git_config(project_root: Path) -> bool:
"""仓库级豁免:git config codeguard.skipGate true。

Expand Down Expand Up @@ -761,11 +816,12 @@ def record_gate_decision(project_root: Path, lang: str, cmd: list[str], rc: int,
path.write_text("\n".join(lines[-limit:]) + "\n", encoding="utf-8")


def record_skip_event(kind: str, project_root: Path | None = None) -> None:
def record_skip_event(kind: str, project_root: Path | None = None,
detail: str | None = None) -> None:
"""记录一次绕过(skipGate/逃生门)到会话状态,Stop 汇总时可见。

除计数外保留最近 20 条明细(时间 + 仓库 + 类型)——豁免必须可回溯:
只有总数时无法回答"哪个仓、什么时候被跳过的"。
除计数外保留最近 20 条明细(时间 + 仓库 + 类型 + 可选 detail)——豁免
必须可回溯:只有总数时无法回答"哪个仓、什么时候、为什么被跳过的"。
"""
state = {}
path = session_state_path()
Expand All @@ -783,6 +839,7 @@ def record_skip_event(kind: str, project_root: Path | None = None) -> None:
"ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
"kind": kind,
"repo": project_root.name if project_root else None,
**({"detail": detail} if detail else {}),
})
del events[:-20]
try:
Expand Down Expand Up @@ -861,6 +918,8 @@ def _failure_detail_blocks(failures: list, *, fix_first: bool = False) -> list[s
f" ▶ 怎么修: {fix}",
f" ▶ 未安装工具时先安装: {hint}",
]
if lang == "python" and "[*]" in (detail or ""):
tail.append(" ▶ 含 [*] 可自动修复项: ruff check --fix <涉及路径>")
body = detail if detail else " lint 退出码非零,无文本输出"
if fix_first:
block += tail + [body]
Expand All @@ -886,19 +945,24 @@ def format_failure_report(failures: list) -> str:
return "\n".join(lines)


def gate_directive(failures: list) -> str:
def gate_directive(failures: list, project_root: Path | str | None = None) -> str:
"""给 AI 的行动指令:收到后应立即修复并重新提交,而不是询问用户。

project_root:本次门禁的目标仓库根——cwd 漂移会让钩子扫到"非目标仓"
(实测:在 codex 仓 cwd 下提交 opencli,POM 4.1.0 报错让人以为改错了
文件),首行显式标注目标仓库可第一时间发现扫错对象。

结构 = 综述(首行契约)→ 强制指令(前置!)→ 每语言细节(fix 先于
detail)→ 修复入口;整体压到 REPORT_MAX_CHARS 内。指令必须前置:宿主把
超长 stderr 从尾部截断,此前指令在报告末段,长报告下 AI 只看到首段 linter
报错、「拆两次调用」与 skipGate 逃生门全部丢失(实测)。首行仍为综述、
综述不重复、含「具体问题」——run_all/硬门禁契约保持。
"""
head_repo = f"本次门禁目标仓库: {Path(project_root).resolve()}\n" if project_root else ""
header = [
summarize_failures(failures),
f"codeguard v{CODEGUARD_VERSION}",
"─" * 60,
head_repo + "─" * 60,
"**给 AI 的强制指令**:提交门禁未通过,禁止执行 git commit / git push。\n"
+ "**⚠️ 整个工具调用没有执行**:被拦截的是一次包含 git commit/push 的完整 Bash "
"调用——其中非 git 的前序步骤(写文件、跑脚本)也全部未运行。请把「修复」与"
Expand All @@ -914,7 +978,8 @@ def gate_directive(failures: list) -> str:
+ "确需绕过(仅用户明确要求时):**单次豁免**用 `git -c codeguard.skipGate=true commit …`"
"(不落配置、无残留,推荐);**仓库级豁免**在该仓库执行 git config codeguard.skipGate true,"
"完成后 git config --unset codeguard.skipGate 恢复。环境变量 CODEGUARD_SKIP_GATE "
"只对手动直调 run_check 有效(无法传入宿主钩子进程)。两种豁免都会记入会话审计明细。",
"只对手动直调 run_check 有效(无法传入宿主钩子进程)。两种豁免都会记入会话审计明细。"
"注意:仓库级豁免对该克隆**所有分支**生效且跨会话残留,务必按上方说明 unset 恢复。",
"─" * 60,
]
footer = f"一键尝试自动修复: python3 {PLUGIN_ROOT}/scripts/fix.py"
Expand Down
27 changes: 25 additions & 2 deletions hooks/pre_tool_git_guard.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
should_suppress_event,
skip_gate_via_git_config, # 规范实现已上移 gate_lib(UPS 也要用)
)
from scope import changed_files

# 拦截的 git 子命令(避免误拦 git status/diff/log 等只读命令)
GUARDED_PATTERNS = ("git commit", "git push")
Expand Down Expand Up @@ -567,6 +568,20 @@ def is_guarded(command: str) -> bool:
return _guarded_mode(command) is not None


def _filter_fallback_roots(roots: list[Path], mode: str,
lanes: tuple[str, ...] | list[str] | None) -> list[Path]:
"""monorepo 兜底面收窄:只保留本次提交面非空的仓。

会话实测:openclaw 提交时,兜底把 workspace 下所有子仓连同根上散落脚本
一起纳入检测面——无关仓的存量违规也变成拦路面(误伤与提交完全无关的
仓库)。无提交面的仓跳过;全部为空返回 [](调用方审计后放行)。

lanes-only:兜底场景没有显式 cd,`git add <相对路径>` 的 extra 属于
会话根而非任何子仓,不参与归属判定。
"""
return [r for r in roots if changed_files(r, mode=mode, lanes=lanes)]


def main() -> int:
ensure_user_path()
payload = read_payload()
Expand Down Expand Up @@ -608,8 +623,16 @@ def main() -> int:
)
return 2
if fallback_note:
# 兜底面收窄(会话实测误拦:无关仓存量违规连带拦提交)。全空 = 本次
# 无可拦对象,审计后放行。
before = len(roots)
roots = _filter_fallback_roots(roots, mode, None)
fallback_note += f";提交面收窄 {before}→{len(roots)} 个仓"
if not roots:
record_skip_event("monorepo-fallback-empty", detail=fallback_note)
return 0
# 兜底走通:仅在会话状态记录(不进 stderr,避免噪音),Stop 摘要可见
record_skip_event("monorepo-fallback", roots[0])
record_skip_event("monorepo-fallback", roots[0], detail=fallback_note)
if any(skip_gate_via_git_config(r) for r in roots):
record_skip_event("skipGate", roots[0])
return 0
Expand Down Expand Up @@ -652,7 +675,7 @@ def main() -> int:
if failures:
# 版本自标识由 gate_directive 首行综述之后的第二行承担——
# 此处不再前置横幅:stderr 首行必须是综述(三个契约测试锁定)。
reports.append(gate_directive(failures))
reports.append(gate_directive(failures, project_root=project_root))
unknown = [s for s in _skipped if "本次改动未涉及" not in s and " SKIPPED:" not in s
and "markdown 风格告警" not in s]
if unknown:
Expand Down
4 changes: 4 additions & 0 deletions scripts/java_project.py
Original file line number Diff line number Diff line change
Expand Up @@ -236,6 +236,10 @@ def analyze(project_root: str | Path, changed: list[str] | None = None) -> dict:
executable = "./" + wrapper if (root / wrapper).is_file() else ("mvn" if system == "maven" else "gradle")
if (root / wrapper).exists() and os.name != "nt" and not os.access(root / wrapper, os.X_OK):
raise ValueError(f"wrapper 不可执行: {wrapper}")
# 顶层 executable 键:门禁消费者(gate_lib Java 门禁)直接读取,
# 用于把 PATH 上的 mvn/gradle 替换为项目自带 wrapper——
# POM 4.1.0 等 Maven 4 仓库在 Maven 3 下必然解析失败(实测)。
plan["executable"] = executable
relevant = None if changed is None else [p for p in changed if p.endswith((".java", ".kt", ".groovy", ".scala"))
or "/src/" in "/" + p
or Path(p).name in _BUILD_FILES or p.startswith((".mvn/", "gradle/"))]
Expand Down
13 changes: 11 additions & 2 deletions scripts/run_per_language.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,9 +154,18 @@ def run_fix(languages: list[str], project_root: Path,
lang_files: list[str] | None = None
if files is not None:
lang_files = [f for f in files if detect_language(f, project_root) == lang]
# .zsh 剥离(与门禁同一份认知):shfmt 只支持 POSIX shell/bash,
# 对 zsh 文件 -w 会重排方言语法造成损坏;跳过并明示修复指令。
zsh_files = [f for f in lang_files if f.endswith(".zsh")]
if zsh_files:
lang_files = [f for f in lang_files if not f.endswith(".zsh")]
results.append({"language": lang, "fixed": False, "skipped": True,
"status": "SKIPPED", "exit_code": 0,
"note": (f"{len(zsh_files)} 个 zsh 文件跳过 formatter"
"(shfmt 不支持 zsh;文件头添加"
" `# shellcheck shell=bash` 注释后"
" shellcheck 门禁即可正常送检)")})
if not lang_files:
results.append({"language": lang, "fixed": False, "skipped": True, "status": "SKIPPED",
"note": "本次改动未涉及该语言"})
continue
fmt = cmd_def.get("format")
if files is not None and not cmd_def.get("append_files", True):
Expand Down
39 changes: 35 additions & 4 deletions scripts/scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,31 @@
)


def is_git_repo(root: str | Path) -> bool:
"""路径自身是否为 git 仓(.git 存在)。"""
return (Path(root) / ".git").exists()


def child_git_repo_names(root: str | Path) -> list[str]:
"""扫描根**不是** git 仓(典型:会话工作区根)时,列出其下自带 .git
的直接子目录名。

这些子目录是独立仓库、由它们各自的提交门禁负责;非 git 根的全量扫描
若把它们一并扫进去,就会出现"提交 A 仓却被 B 仓的存量问题拦截"的
跨仓误伤(2026-09-23 opencli-java-sdk 会话实测:workspace 根的
push-all-branches.sh 触发门禁时,根级脚本与子仓文件混在同一份报告里)。
根级别的散文件(不属于任何子仓)仍保留在扫描面内——它们没有别的门禁。
"""
out: list[str] = []
try:
for child in sorted(Path(root).iterdir()):
if child.is_dir() and (child / ".git").exists():
out.append(child.name)
except OSError:
pass
return out


def is_build_artifact(path: str | Path) -> bool:
"""路径是否落在构建产物/依赖快照目录下(任一段命中即算)。

Expand All @@ -63,7 +88,7 @@ def is_build_artifact(path: str | Path) -> bool:
_RUFF_SNIPPET = Path(__file__).resolve().parents[1] / "linters" / "ruff" / "ruff.toml"


def _inject_find_excludes(expr: str) -> str:
def _inject_find_excludes(expr: str, excludes: tuple[str, ...] | list[str] = FULL_SCAN_EXCLUDES) -> str:
"""给 `find …` 型 gate 表达式注入构建产物目录排除(-not -path)。

覆盖三种实测形态——只认一种就有整族门禁漏网:
Expand All @@ -79,7 +104,7 @@ def _inject_find_excludes(expr: str) -> str:
return expr
additions = "".join(
f" -not -path '*/{d}/*'"
for d in FULL_SCAN_EXCLUDES
for d in excludes
if f"'*/{d}/*'" not in expr and f"'*/{d}'" not in expr
)
if not additions:
Expand Down Expand Up @@ -179,16 +204,22 @@ def scope_cmd(
if not out:
return out
targets = [single_file] if single_file else list(files or [])
root = Path(project_root)
# 非 git 根(会话工作区)的全量扫描:排除子 git 仓——它们由各自的
# 提交门禁负责,混进来就是跨仓误伤(见 child_git_repo_names)。
scan_excludes = list(FULL_SCAN_EXCLUDES)
if full_excludes and not is_git_repo(root):
scan_excludes += child_git_repo_names(root)
if "{file}" in " ".join(out):
return [c.replace("{file}", single_file or "") for c in out]
if out[0] == "ruff":
args = ruff_config_args(out, project_root)
out = [out[0]] + args + out[1:]
if full_excludes:
for d in FULL_SCAN_EXCLUDES:
for d in scan_excludes:
out += ["--exclude", d]
elif full_excludes:
out = [_inject_find_excludes(c) if isinstance(c, str) else c for c in out]
out = [_inject_find_excludes(c, scan_excludes) if isinstance(c, str) else c for c in out]
scan_idx = [i for i, tok in enumerate(out[1:], 1) if tok == "." or "**" in tok]
if targets and scan_idx:
flags = [tok for i, tok in enumerate(out) if i not in scan_idx and not tok.startswith("#")]
Expand Down
Loading
Loading