From 8e2157d69862e95ed706213e22e386bf4758e40f Mon Sep 17 00:00:00 2001 From: loong10k <20489781+loong10k@users.noreply.github.com> Date: Wed, 23 Sep 2026 13:09:25 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E6=94=B6=E6=95=9B=E4=BA=94?= =?UTF-8?q?=E7=B1=BB=20CodeGuard=20Hook=20=E5=BA=94=E7=94=A8=E5=B1=82=20(v?= =?UTF-8?q?0.14.0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .agents/plugins/marketplace.json | 8 +- .codex-plugin/plugin.json | 2 +- .zcode-plugin/plugin.json | 2 +- docs/current-architecture.md | 8 +- hooks/env_check.py | 215 +++---------- hooks/post_tool_lint.py | 292 +++--------------- hooks/pre_tool_git_guard.py | 191 ++---------- hooks/stop_summary.py | 52 +--- hooks/user_prompt_validator.py | 246 +++------------ kimi.plugin.json | 2 +- .../refactor-codeguard-architecture/tasks.md | 4 +- .../verification.md | 15 + scripts/check_architecture.py | 22 +- scripts/codeguard/git_context.py | 8 +- scripts/codeguard/git_guard_application.py | 128 ++++++++ scripts/codeguard/prompt_application.py | 106 +++++++ scripts/codeguard/prompt_policy.py | 56 ++++ scripts/codeguard/save_application.py | 201 ++++++++++++ scripts/codeguard/session_application.py | 60 ++++ scripts/codeguard/startup_application.py | 130 ++++++++ tests/run_all.py | 7 +- tests/test_architecture.py | 45 +++ tests/test_check_plans.py | 4 +- tests/test_git_boundaries.py | 10 + tests/test_git_guard_application.py | 74 +++++ tests/test_prompt_application.py | 67 ++++ tests/test_save_application.py | 77 +++++ tests/test_session_application.py | 56 ++++ tests/test_startup_application.py | 48 +++ tests/test_verdict_integrity.py | 11 +- 30 files changed, 1300 insertions(+), 847 deletions(-) create mode 100644 scripts/codeguard/git_guard_application.py create mode 100644 scripts/codeguard/prompt_application.py create mode 100644 scripts/codeguard/prompt_policy.py create mode 100644 scripts/codeguard/save_application.py create mode 100644 scripts/codeguard/session_application.py create mode 100644 scripts/codeguard/startup_application.py create mode 100644 tests/test_git_guard_application.py create mode 100644 tests/test_prompt_application.py create mode 100644 tests/test_save_application.py create mode 100644 tests/test_session_application.py create mode 100644 tests/test_startup_application.py diff --git a/.agents/plugins/marketplace.json b/.agents/plugins/marketplace.json index a881c6a..372e5be 100644 --- a/.agents/plugins/marketplace.json +++ b/.agents/plugins/marketplace.json @@ -9,20 +9,20 @@ "source": { "source": "url", "url": "https://github.com/partme-ai/partme-codeguard-plugin.git", - "ref": "v0.13.0" + "ref": "v0.14.0" }, "policy": { "installation": "AVAILABLE", "authentication": "ON_USE" }, "category": "Developer Tools", - "version": "0.13.0", + "version": "0.14.0", "description": "Evidence-backed code checks and Git content gates for AI assistants, with Maven/Gradle module impact analysis. Save hooks provide feedback; unverified checks are explicit.", - "icon": "https://cdn.jsdelivr.net/gh/full-stack-plugins/codeguard-plugin@v0.13.0/assets/official-logo.png", + "icon": "https://cdn.jsdelivr.net/gh/full-stack-plugins/codeguard-plugin@v0.14.0/assets/official-logo.png", "interface": { "displayName": "代码规范守卫", "shortDescription": "Trustworthy code checks and Java impact analysis", - "logo": "https://cdn.jsdelivr.net/gh/full-stack-plugins/codeguard-plugin@v0.13.0/assets/official-logo.png" + "logo": "https://cdn.jsdelivr.net/gh/full-stack-plugins/codeguard-plugin@v0.14.0/assets/official-logo.png" } } ] diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index 5b80144..f3160ef 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "codeguard", - "version": "0.13.0+codex.20260923", + "version": "0.14.0+codex.20260923", "description": "Evidence-backed code checks and Git content gates for AI assistants, with Maven/Gradle module impact analysis. Save hooks provide feedback; unverified checks are explicit.", "author": { "name": "Full Stack Skills / PartMe.AI", diff --git a/.zcode-plugin/plugin.json b/.zcode-plugin/plugin.json index 04e54d9..1a96311 100644 --- a/.zcode-plugin/plugin.json +++ b/.zcode-plugin/plugin.json @@ -5,7 +5,7 @@ "en": "CodeGuard", "zh-CN": "代码规范检查" }, - "version": "0.13.0", + "version": "0.14.0", "description": "Evidence-backed code checks and Git content gates for AI assistants, with Maven/Gradle module impact analysis. Save hooks provide feedback; unverified checks are explicit.", "description_i18n": { "en": "Evidence-backed code checks and Git content gates for AI assistants, with Maven/Gradle module impact analysis. Save hooks provide feedback; unverified checks are explicit.", diff --git a/docs/current-architecture.md b/docs/current-architecture.md index 647af9f..13844ff 100644 --- a/docs/current-architecture.md +++ b/docs/current-architecture.md @@ -7,7 +7,10 @@ ```mermaid flowchart TD Host[Codex / ZCode / Kimi] --> Entry[CLI / MCP / 五类 Hook] + Entry --> HookApp[startup / session / prompt / git_guard / save applications] Entry --> Check[check_application / language_check] + HookApp --> Gate + HookApp --> Plan Entry --> Gate[gate / gate_checks] Entry --> Scan[cve / dockerfile] Check --> Plan[planning / Java 模块分析] @@ -30,8 +33,9 @@ flowchart TD | 位置 | 责任 | 扩展边界 | |---|---|---| | `bin/codeguard`、`scripts/run_check.py`、`scripts/fix.py` | CLI 参数、输出与 MCP stdio 注册 | 不复制扫描器或判定逻辑 | -| `hooks/` | 宿主 JSON、会话事件、通知与退出协议 | PreToolUse/PostToolUse 仍有待迁移的编排，不能称为纯适配器 | +| `hooks/` | 宿主 JSON、会话事件、通知与退出协议 | 五类 Hook 将检查编排委托应用服务；入口保留旧导入兼容面及 fail-open 协议 | | `scripts/codeguard/check_application.py`、`language_check.py` | 语言选择、检查与修复请求、MCP 工具分发 | 不导入 MCP SDK；旧 `run_per_language.py` 仅再导出 | +| `startup_application.py`、`session_application.py`、`prompt_application.py`、`git_guard_application.py`、`save_application.py` | 五类 Hook 的盘点、状态消费、软提示、Git 守卫及保存反馈编排 | 返回结构化结果，不打印或调用宿主通知；Hook 控制输出、通知与 fail-open。软/硬提示与保存统计在成功输出后确认；Stop 的状态消费仍先于 stdout，输出故障时可能丢失总结 | | `gate.py`、`gate_checks.py`、`repository_policy.py`、`baseline.py` | Git 门禁、单语言结果、安全规则及有证据的存量比较 | 只有准确基线失败且每条诊断及次数覆盖当前结果才可豁免 | | `java_build.py`、`java_impact.py`、`java_planning.py` 等 | 构建读取、纯影响闭包、命令选择与环境观察 | 默认保留跳过测试的行为；复杂构建保守扩大检查范围 | | `cve_reports.py`、`cve_policy.py`、`cve_scanners.py`、`cve.py` | 漏洞报告、阈值、进程适配与复扫编排 | 无有效结构化报告就没有安全通过结论 | @@ -62,4 +66,4 @@ python3 scripts/vendor/skill_vendor.py check openspec validate refactor-codeguard-architecture --strict ``` -真实 Maven/Gradle 大型项目、联网漏洞数据库、57 种工具链、Windows 与三宿主已安装运行均需要独立证据。临时 Git 快照不是执行沙箱；构建和扫描命令以当前用户权限运行。当前重构的任务进度和本地证据见 `openspec/changes/refactor-codeguard-architecture/verification.md`，发布记录需待版本、CI、tag、市场和宿主分别核对。 +真实 Maven/Gradle 大型项目、联网漏洞数据库、57 种工具链、Windows 与三宿主已安装运行均需要独立证据。临时 Git 快照不是执行沙箱；构建和扫描命令以当前用户权限运行。当前重构的任务进度和本地证据见 `openspec/changes/refactor-codeguard-architecture/verification.md`。v0.13.0 的源码、CI、tag、Release 和市场已核对；其后工作树变更仍需独立发布，宿主安装运行尚未验收。 diff --git a/hooks/env_check.py b/hooks/env_check.py index b0ff428..bfc1585 100755 --- a/hooks/env_check.py +++ b/hooks/env_check.py @@ -1,11 +1,7 @@ #!/usr/bin/env python3 -"""SessionStart 钩子：检测项目语言，把规则注入 AI 上下文。 +"""SessionStart 宿主适配器：读取事件与本机插件配置，输出项目记忆。""" +from __future__ import annotations -行为： -1. detect_lang 找到项目语言 -2. 检查是否已有 linter 配置（.pre-commit-config.yaml、checkstyle.xml、.clippy.toml、eslint.config.js、ruff.toml） -3. 输出一段 AGENTS.md 风格的提示，让 AI 知道自己在一个被 codeguard 管理的项目里 -""" import contextlib import json import os @@ -13,10 +9,12 @@ import sys from pathlib import Path -PLUGIN_ROOT = Path(__file__).resolve().parents[1] # hooks/ 的上级 = 插件根（不依赖宿主环境变量） +PLUGIN_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PLUGIN_ROOT / "scripts")) -from detect_lang import ( # ensure_user_path/load_user_config 实际定义：scripts/paths.py、scripts/user_config.py +from codeguard import startup_application +from codeguard.reporting import plugin_version +from detect_lang import ( # 兼容原有 Python 导入接口 LANG_COMMANDS, REGISTRY, detect_languages, @@ -25,79 +23,49 @@ load_user_config, ) +__all__ = [ + "LANG_COMMANDS", + "REGISTRY", + "detect_languages", + "detect_linter_config", + "ensure_user_path", + "find_project_root", + "load_user_config", + "main", + "notify", + "version_backlog_note", +] + +detect_linter_config = startup_application.detect_linter_config +version_backlog_note = startup_application.version_backlog_note + def notify(title: str, message: str) -> None: - """macOS 系统通知（非 darwin 静默）""" + """macOS 系统通知；其他宿主静默。""" if sys.platform != "darwin": return - safe_t = title.replace('"', "'") - safe_m = message.replace('"', "'")[:200] + safe_title = title.replace('"', "'") + safe_message = message.replace('"', "'")[:200] with contextlib.suppress(OSError): subprocess.Popen( ["osascript", "-e", - f'display notification "{safe_m}" with title "{safe_t}"'], + f'display notification "{safe_message}" with title "{safe_title}"'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) -def detect_linter_config(project_root: Path) -> dict: - """返回 {language: [存在的 linter 配置文件]}，由 languages.json 驱动。 - - 每条 languages.json 条目的 `linter_config_files` 字段声明该语言的 linter - 配置文件路径（glob 模式允许，与 requiresConfig 同语义）。新增语言只需 - 在注册表里填字段，env_check 自动盘点——不再有硬编码常量的同步陷阱。 - """ - found = {} - for lang_id, lang_def in REGISTRY.items(): - files = lang_def.get("linter_config_files") or [] - if not files: - continue - present = [f for f in files if (project_root / f).exists()] - if present: - found[lang_id] = present - return found - - -def _semver_tuple(v: str) -> tuple[int, ...]: - import re as _re - m = _re.match(r"(\d+(?:\.\d+)*)", v.strip()) - if not m: - return (0,) - return tuple(int(x) for x in m.group(1).split(".")) - - -def version_backlog_note(current: str, cache_root: Path) -> str | None: - """纯函数：本地插件 cache 存在比 current 更新的版本 → 提示文案；否则 None。 - - 检测面 = `~/.zcode/cli/plugins/cache//codeguard/` 的**版本目录 - 集合**（离线可得的唯一权威）。抓两类漂移：同 vendor 升级后旧目录未清、跨 - vendor 版本参差导致实际生效副本落后于本机已有副本。抓不到"本机最新 vs - 远端仓库最新"（需网络，钩子不联网）——文案里明示提醒核对远端。 - """ - newer: list[str] = [] - installed: list[str] = [] - if not cache_root.is_dir(): - return None +def _enabled_plugin_orgs() -> tuple[str, ...]: + """ZCode 宿主注册表只在适配器读取，其他宿主缺文件时无副作用。""" try: - for vendor_dir in sorted(cache_root.glob("*/codeguard/*")): - if not vendor_dir.is_dir(): - continue - v = vendor_dir.name - installed.append(v) - if _semver_tuple(v) > _semver_tuple(current): - newer.append(v) - except OSError: - return None - if not newer: - return None - newest = max(newer, key=_semver_tuple) - return ( - f"- ⚠️ **版本积压**：当前生效 codeguard `v{current}`，本机 cache 已有更新版本 " - f"`v{newest}`（共 {len(installed)} 份: {', '.join(sorted(installed))}）——" - "钩子行为以**生效副本**为准，修复可能已存在却没跑到；请更新插件并清理旧 " - "cache（`~/.zcode/cli/plugins/cache/*/codeguard/`），同时核对远端是否还有" - "更新版本（本检测离线，看不到仓库最新版）。" - ) + config = json.loads((Path.home() / ".zcode" / "cli" / "config.json").read_text(encoding="utf-8")) + orgs = { + key.split("@", 1)[1] + for key in (config.get("plugins", {}) or {}).get("enabledPlugins", {}) + if key.startswith("codeguard@") + } + return tuple(sorted(orgs)) + except (OSError, ValueError, AttributeError): + return () def main(payload: dict | None = None) -> int: @@ -108,100 +76,19 @@ def main(payload: dict | None = None) -> int: def _main(payload: dict | None = None) -> int: - ensure_user_path(from_login_shell=True) # GUI 宿主 PATH 不含用户级工具目录，先补齐再盘点 - project_root = find_project_root(os.getcwd()) - if project_root is None: - project_root = Path(os.getcwd()) - - languages = detect_languages(project_root) - if not languages: - return 0 # 未识别到代码项目，安静退出 - - linter_cfg = detect_linter_config(project_root) - cfg = load_user_config() - enabled = cfg.get("enabled_languages", []) - if enabled and enabled != ["auto"]: - languages = [lang for lang in languages if lang in enabled] - - # linter 盘点：直接复用门禁的探活（比裸 which 准——能发现 npx 包未装、运行时损坏） - from codeguard.toolchain import ToolchainProbe - probe = ToolchainProbe(project_root) - ready, missing = [], [] - for lang in languages: - cmd_def = LANG_COMMANDS.get(lang) - if not cmd_def: - continue - ok, reason = probe.probe(cmd_def) - if ok: - ready.append(lang) - else: - hint = cmd_def.get("install_hint") or "见 docs/LANGUAGES.md" - missing.append((lang, reason, hint)) - - # 输出注入到 AI 上下文的内容（追加模式） - lines = [] - lines.append("# codeguard 项目记忆") - lines.append("") - lines.append(f"- 项目根: `{project_root}`") - lines.append(f"- 检测到语言: {', '.join(languages)}") - - if missing: - lines.append("") - lines.append("**⚠️ 以下语言的检查未生效，这些语言的文件【不会】被检查（提交门禁也会跳过它们）：**") - for lang, reason, hint in missing: - lines.append(f" - {lang}（{reason}）→ 修复: `{hint}`") - lines.append(" 请主动提示用户安装；安装后新写的文件才会被检查。") - notify( - "codeguard 部分检查未生效", - f"{len(missing)} 个语言未生效: " + ", ".join(lang for lang, _, _ in missing) + - "；详情见对话", - ) - if ready: - lines.append(f"- 检查已生效的语言: {', '.join(ready)}") - - if linter_cfg: - lines.append(f"- 已有的 linter 配置: {linter_cfg}") - else: - lines.append("- 未配置任何 linter（建议运行 /init 接入 codeguard）") - - # 双副本检测：partme-ai 与 full-stack-plugins 同时启用时每个钩子事件 - # 跑两遍（历史实测双份触发）；提示用户停用其一，而不是默默重复。 - try: - import json as _json - _reg = _json.loads((Path.home() / ".zcode" / "cli" / "config.json").read_text(encoding="utf-8")) - _orgs = sorted({ - key.split("@", 1)[1] - for key in (_reg.get("plugins", {}) or {}).get("enabledPlugins", {}) - if key.startswith("codeguard@") - }) - except (OSError, ValueError, AttributeError): - _orgs = [] - if len(_orgs) > 1: - lines.append( - f"- ⚠️ 检测到 codeguard **双副本同时启用**（{', '.join(_orgs)}）：" - "每个钩子事件会执行两遍，报告与统计可能翻倍或分裂——建议只保留一个来源" - ) - - # 版本积压：本机 cache 里有比当前生效副本更新的版本（实测 8 份副本停留在 - # 0.5.x 而源码已 0.8.x——"发布了但没跑到"的盲区，SessionStart 即可见） - try: - _mf = json.loads((PLUGIN_ROOT / ".zcode-plugin" / "plugin.json").read_text(encoding="utf-8")) - _current = str(_mf.get("version") or "") - except (OSError, ValueError, AttributeError): - _current = "" - if _current: - _note = version_backlog_note( - _current, Path.home() / ".zcode" / "cli" / "plugins" / "cache" - ) - if _note: - lines.append(_note) - - lines.append("- AI 写完代码会被 PostToolUse 钩子自动 lint，告警会出现在这里，按告警里的「怎么修」处理") - lines.append("- 用户要求「提交/push」时，UserPromptSubmit 钩子会再次确认所有 linter 通过，未通过会拦截提交") - lines.append("") - lines.append("**重要**：写代码前先阅读对应语言的规范（见 skills/codeguard-{language}/SKILL.md）") - - print("\n".join(lines)) + ensure_user_path(from_login_shell=True) + project_root = find_project_root(os.getcwd()) or Path(os.getcwd()) + current = plugin_version() + report = startup_application.build_startup_report( + project_root, + plugin_version="" if current == "unknown" else current, + cache_root=Path.home() / ".zcode" / "cli" / "plugins" / "cache", + enabled_plugin_orgs=_enabled_plugin_orgs(), + ) + if report.notification: + notify(*report.notification) + if report.text: + print(report.text) return 0 @@ -218,6 +105,6 @@ def _main(payload: dict | None = None) -> int: sys.exit(main(payload)) except SystemExit: raise - except Exception as exc: # noqa: BLE001 — 内部错误 fail-open：traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open，不阻断宿主工作流 print(f"[codeguard] 内部错误已忽略（fail-open）: {exc!r}", file=sys.stderr) sys.exit(0) diff --git a/hooks/post_tool_lint.py b/hooks/post_tool_lint.py index 1d91abb..f72ed40 100644 --- a/hooks/post_tool_lint.py +++ b/hooks/post_tool_lint.py @@ -1,12 +1,4 @@ -"""codeguard PostToolUse 钩子：AI 写文件后自动 lint 并注入告警到 AI 上下文。 - -输出协议（三端兼容，均为 stdout 合法 JSON、exit 0 不阻断）： - ZCode → 解析 hookSpecificOutput.additionalContext 注入 AI 上下文 - Codex CLI → 同一协议注入 developer context；systemMessage 显示为 UI 警告 - （纯文本 stdout 会被忽略，必须 JSON） - Kimi Code → PostToolUse 观察型，exit 0 时 stdout 内容附加到上下文 - 用户同时收到 macOS 系统通知（可关闭）。 -""" +"""PostToolUse 宿主适配器：保存反馈以 JSON 注入上下文，绝不阻断写文件。""" from __future__ import annotations import json @@ -15,15 +7,12 @@ import sys from pathlib import Path -PLUGIN_ROOT = Path(__file__).resolve().parents[1] # hooks/ 的上级 = 插件根 +PLUGIN_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PLUGIN_ROOT / "scripts")) -from codeguard import hook_state -from codeguard.execution import execute -from codeguard.fingerprint import check_identity -from codeguard.hook_state import codeguard_home -from codeguard.planning import scoped_plan -from detect_lang import ( # ensure_user_path/load_user_config 实际定义：scripts/paths.py、scripts/user_config.py +from codeguard import save_application +from codeguard.hook_state import codeguard_home, session_scope +from detect_lang import ( # 兼容原有 Python 导入接口 LANG_COMMANDS, detect_language, ensure_user_path, @@ -33,9 +22,14 @@ ) from scope import is_build_artifact -# 双副本去重：同一插件可能以多个 marketplace 副本安装（partme-ai/ 与 -# full-stack-plugins/ 各一份，钩子双份触发——实测），用户级固定路径跨副本共享 -# 路径按调用时的 CODEGUARD_HOME 解析；保留覆盖点供嵌入调用者隔离存储。 +__all__ = [ + "DEDUP_FILE", "LANG_COMMANDS", "bump_state", "detect_language", "ensure_user_path", + "extract_file_path", "find_project_root", "is_build_artifact", "is_ejs_template", + "load_user_config", "mac_notify", "main", "project_uses_linter", "read_payload", + "run", "should_skip", "should_suppress_duplicate", +] + +# 多个 marketplace 副本共享用户级去重状态；保留原测试/嵌入方的覆盖点。 DEDUP_FILE: Path | None = None @@ -44,48 +38,37 @@ def _dedup_path() -> Path: def _state_path() -> Path: - """旧无 ID 状态可回退；带会话 ID 的新统计绝不认领无归属旧数据。""" - return hook_state.state_path(PLUGIN_ROOT / ".session_state.json") + return save_application.state_path() def bump_state(lang: str, passed: bool, auto_fixed: bool = False) -> None: - hook_state.bump_state(_state_path(), lang, passed, auto_fixed) + save_application.bump_state(lang, passed, auto_fixed) def _notify_cooldown_ok(lang: str, cooldown_s: int = 60) -> bool: - return hook_state.notification_allowed(_state_path(), lang, cooldown_s) + return save_application.notification_allowed(lang, cooldown_s) + def mac_notify(title: str, message: str) -> None: - """macOS 系统通知（用户视觉提醒；非 macOS 静默跳过）""" + """macOS 通知；非 macOS 宿主由 JSON 上下文呈现问题。""" if sys.platform != "darwin": return safe_title = title.replace('"', "'") - safe_msg = message.replace('"', "'")[:200] + safe_message = message.replace('"', "'")[:200] subprocess.Popen( ["osascript", "-e", - f'display notification "{safe_msg}" with title "{safe_title}" sound name "Pop"'], + f'display notification "{safe_message}" with title "{safe_title}" sound name "Pop"'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def should_suppress_duplicate(file_path: str, identity: str | None = None) -> bool: - return hook_state.completed_file_check(file_path, _dedup_path(), identity) - - -def is_ejs_template(file_path: str) -> bool: - """vue-cli 的 public/index.html 是 EJS 模板（<%= %>/<% if %>）。htmlhint - 对 EJS 标记必然误报（spec-char-escape / attr-value-double-quotes）， - 且 CLI 不支持按规则覆盖（--rule=false 报 unknown option，实测）—— - 这类文件直接跳过：EJS 语法错误由 vue-cli 构建流程兜底。""" - try: - return "<%" in Path(file_path).read_text(errors="ignore")[:8192] - except OSError: - return False + return save_application.should_suppress_duplicate(file_path, identity, _dedup_path()) -def run(cmd: list[str], cwd: Path, timeout: int = 300) -> tuple[int, str, str]: - """保存钩子的兼容执行入口，不拥有独立的故障判定。""" - return execute(cmd, cwd, timeout).as_tuple() +is_ejs_template = save_application.is_ejs_template +run = save_application.run +should_skip = save_application.should_skip def read_payload() -> dict: @@ -98,227 +81,42 @@ def read_payload() -> dict: def extract_file_path(payload: dict) -> str: - for c in (payload.get("file_path"), - (payload.get("tool_input") or {}).get("file_path"), - (payload.get("tool_input") or {}).get("path"), - (payload.get("input") or {}).get("file_path")): - if c: - return str(c) + for candidate in (payload.get("file_path"), + (payload.get("tool_input") or {}).get("file_path"), + (payload.get("tool_input") or {}).get("path"), + (payload.get("input") or {}).get("file_path")): + if candidate: + return str(candidate) return "" -def should_skip(file_path: str, languages: list[str]) -> tuple[bool, str]: - if not file_path: - return True, "" - # 构建产物静默跳过：写到 target/site、target/apidocs 的生成 HTML/sh 由 - # 构建流程负责，检查它们只会给出下次构建就被重写的假告警，AI 还可能 - # "自动修复"生成物（prettier 改完、构建一跑又变回去）。 - # 目录认知单源：scope.FULL_SCAN_EXCLUDES。 - if is_build_artifact(file_path): - return True, "" - lang = detect_language(file_path) - if not lang: - return True, "" - if languages and lang not in languages: - return True, lang - return False, lang - - def main() -> int: - from codeguard.hook_state import session_scope payload = read_payload() with session_scope(payload): return _main(payload) def _main(payload: dict) -> int: - ensure_user_path() # 高频钩子：仅补静态目录（零开销），linter 找得到才跑得起来 + ensure_user_path() file_path = extract_file_path(payload) if not file_path: return 0 - cfg = load_user_config() - enabled = cfg.get("enabled_languages", []) project_root = find_project_root(os.getcwd()) or Path(os.getcwd()) - - skip, lang = should_skip(file_path, enabled) - if skip: - return 0 - - # 文档类语言不拦 AI 写作流（AI 产出的报告/文档常不合 lint 严格规则， - # 拦截会造成死循环）；提交门禁阶段仍有宽松校验 - if lang == "markdown": - return 0 - - cmd_def = LANG_COMMANDS.get(lang) - if not cmd_def: - return 0 - - if not cmd_def.get("append_files", True): - print(json.dumps({"hookSpecificOutput": { - "hookEventName": "PostToolUse", - "additionalContext": f"codeguard: {lang} 是项目级检查；本次保存未验证，" - "请使用 codeguard check 或提交门禁。不会自动格式化整个项目。", - }}, ensure_ascii=False)) - return 0 - - # 项目未接入该 linter（无配置文件）时，生态型 linter（eslint）必然报错 - # 退出——那是「未接入」不是「代码违规」（qumall-mall-ui 无 .eslintrc 实测） - if not project_uses_linter(cmd_def, project_root): - return 0 - - # EJS 模板（vue-cli public/index.html）直接跳过 htmlhint：CLI 不支持 - # 按规则豁免（实测 unknown option），而误报是必然的 - if lang == "html" and is_ejs_template(file_path): - return 0 - lint_cmd = cmd_def.get("lint") or [] - if not lint_cmd: - return 0 - - timeout = cfg.get("lint_timeout_seconds", 120) - - # 保存范围显式为 save；先冻结检查命令，修复后原样复检。 - lint_plan = scoped_plan(lint_cmd, project_root, scope="save", files=[str(file_path)]) - lint_argv = list(lint_plan.commands[0].argv) - def identity(): - return check_identity(project_root, cfg, {lang: cmd_def}, extra_files=[file_path]) - checked_identity = identity() - if checked_identity is not None and should_suppress_duplicate(file_path, checked_identity): - return 0 - rc, stdout, stderr = run(lint_argv, cwd=lint_plan.cwd, timeout=timeout) - - from verdict import UNVERIFIED, lint_verdict - verdict, reason = lint_verdict(rc, lint_cmd, stdout + stderr) - if verdict == UNVERIFIED: - print(json.dumps({"hookSpecificOutput": { - "hookEventName": "PostToolUse", - "additionalContext": f"codeguard: UNVERIFIED [{lang}] {reason} (exit {rc})；" - "没有代码结论，不自动修复。", - }}, ensure_ascii=False)) - return 0 - - if rc == 0: - if checked_identity is not None and checked_identity == identity(): - hook_state.record_file_check(file_path, _dedup_path(), checked_identity) - bump_state(lang, passed=True) - # 注入 AI 上下文：告诉 AI 该文件通过了门禁 - print(json.dumps({ - "hookSpecificOutput": { - "hookEventName": "PostToolUse", - "additionalContext": f"codeguard: ✅ {lang} lint passed for {Path(file_path).name}" - }, - "systemMessage": f"codeguard: ✅ {lang} lint passed", - }, ensure_ascii=False)) - return 0 - - # lint 失败 → 尝试自动修复 - auto_fixed = False - touched_others: list[str] = [] - if cfg.get("auto_fix_on_save", True): - fmt_cmd = cmd_def.get("format") - if fmt_cmd: - - def _porcelain() -> set[str]: - try: - import subprocess as _sp - proc = _sp.run( - ["git", "status", "--porcelain"], - cwd=project_root, capture_output=True, check=False, - text=True, timeout=10, - ) - return { - ln[3:].strip() for ln in proc.stdout.splitlines() - if len(ln) > 3 - } if proc.returncode == 0 else set() - except Exception: # noqa: BLE001 — 快照失败仅失去清单注入，不影响修复 - return set() - - before = _porcelain() - fix_plan = scoped_plan(fmt_cmd, project_root, scope="save", files=[str(file_path)]) - frc, _, _ = run(list(fix_plan.commands[0].argv), cwd=fix_plan.cwd, timeout=timeout + 60) - if frc == 0: - auto_fixed = True - after = _porcelain() - touched_others = sorted( - name for name in (after - before) - if not name.rstrip("/").endswith(Path(file_path).name) - ) - checked_identity = identity() - rc, stdout, stderr = run(lint_argv, cwd=lint_plan.cwd, timeout=timeout) - - verdict, reason = lint_verdict(rc, lint_cmd, stdout + stderr) - if verdict == UNVERIFIED: - print(json.dumps({"hookSpecificOutput": { + result = save_application.evaluate_save(file_path, project_root, cfg, store=_dedup_path()) + if result.additional_context is not None: + message = {"hookSpecificOutput": { "hookEventName": "PostToolUse", - "additionalContext": f"codeguard: UNVERIFIED [{lang}] 修复后复检未完成：{reason}；" - "formatter 可能已修改文件，请重新读取。没有通过/失败结论。", - }}, ensure_ascii=False)) - return 0 - passed = rc == 0 - if checked_identity is not None and checked_identity == identity(): - hook_state.record_file_check(file_path, _dedup_path(), checked_identity) - bump_state(lang, passed=passed, auto_fixed=auto_fixed) - - if passed: - # 已自动修复：告知 AI 文件被工具改过，避免 AI 继续用旧内容 - note = "" - if auto_fixed: - note = f"（已自动运行 {' '.join(cmd_def.get('format') or [])} 修复）" - if touched_others: - shown = ", ".join(touched_others[:10]) - more = f" 等 {len(touched_others)} 个" if len(touched_others) > 10 else "" - note += ( - f"。⚠️ format 命令还改动了本文件之外的文件{more}：{shown}" - "——这些文件你内存里的版本已过期，必须先重新读取再继续操作" - ) - else: - note += "（仅本文件被改动，请重新读取）" - print(json.dumps({ - "hookSpecificOutput": { - "hookEventName": "PostToolUse", - "additionalContext": f"codeguard: ✅ {lang} lint passed for {Path(file_path).name}{note}" - }, - "systemMessage": f"codeguard: ✅ {lang} lint passed", - }, ensure_ascii=False)) - return 0 - - # ===== 结构化告警：AI 直接看到「什么问题 + 怎么修」 ===== - # linter 输出头部是最具体的问题（shellcheck/ruff/eslint 均如此），取头部而非尾部 - raw = (stdout or stderr or "").strip() - problem_lines = [ln.strip() for ln in raw.splitlines() if ln.strip()] - problems = "\n".join(problem_lines[:5])[:500] or "（linter 未输出具体问题）" - if len(problem_lines) > 5 or len(raw) > 500: - # 节选会让 AI 一次修 3 个、重试再看 3 个（whack-a-mole）：给总量+完整日志 - try: - import hashlib as _h - import tempfile as _tf - key = _h.sha1(str(file_path).encode()).hexdigest()[:12] - full_path = Path(_tf.gettempdir()) / f"codeguard-post-{lang}-{key}.log" - full_path.write_text(raw, encoding="utf-8") - problems += f"\n…（截断，共 {len(problem_lines)} 行；完整输出: {full_path}）" - except OSError: - problems += f"\n…（截断，共 {len(problem_lines)} 行）" - fix_cmd = " ".join(cmd_def.get("format") or []) or "按上述问题逐条修复" - context = ( - f"codeguard ⚠️ [{lang}] {Path(file_path).name} 检查未通过\n" - f"发现的问题（节选）:\n{problems}\n" - f"怎么修:\n" - f" 1. 可先运行 `{fix_cmd}` 自动修复格式类问题\n" - f" 2. 手动修复上述具体问题后，重新保存该文件，codeguard 会自动复检" - ) - print(json.dumps({ - "hookSpecificOutput": { - "hookEventName": "PostToolUse", - "additionalContext": context - }, - "systemMessage": f"codeguard: ⚠️ {lang} lint FAILED: {Path(file_path).name}", - }, ensure_ascii=False)) - - # macOS 系统通知（用户视觉提醒；同语言 60s 冷却——批量编辑时不轰炸） - alert = f"{lang} lint FAILED: {Path(file_path).name}" - if _notify_cooldown_ok(lang): - mac_notify(alert, problem_lines[0][:120] if problem_lines else "") - + "additionalContext": result.additional_context, + }} + if result.system_message is not None: + message["systemMessage"] = result.system_message + print(json.dumps(message, ensure_ascii=False)) + result.acknowledge() + if result.notification: + lang, title, message = result.notification + if _notify_cooldown_ok(lang): + mac_notify(title, message) return 0 @@ -327,6 +125,6 @@ def _porcelain() -> set[str]: sys.exit(main()) except SystemExit: raise - except Exception as exc: # noqa: BLE001 — 内部错误 fail-open：traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open，不阻断 AI 写文件 print(f"[codeguard] 内部错误已忽略（fail-open）: {exc!r}", file=sys.stderr) sys.exit(0) diff --git a/hooks/pre_tool_git_guard.py b/hooks/pre_tool_git_guard.py index 66bf79b..7dc145c 100755 --- a/hooks/pre_tool_git_guard.py +++ b/hooks/pre_tool_git_guard.py @@ -1,13 +1,5 @@ #!/usr/bin/env python3 -"""PreToolUse 钩子：AI 执行 git commit / git push 前硬门禁。 - -这是「修完才能推」的硬保证：AI 即使忽略 UserPromptSubmit 的软引导指令， -真正执行 git 命令时也会在这里被拦下。exit 2 + stderr 的反馈会作为工具 -结果回给 AI（Claude/ZCode 协议），AI 看到后继续修复并重试——循环闭环： -失败 → AI 修 → 再 commit → 通过 → 放行。 - -通过时完全静默（exit 0，零输出）。 -""" +"""PreToolUse 宿主适配器：AI 执行 git commit/push 前输出硬门禁决策。""" from __future__ import annotations import contextlib @@ -32,6 +24,7 @@ resolve_project_roots, staging_intent, ) +from codeguard.git_guard_application import evaluate_git_command from codeguard.git_syntax import ( _analyze_segment, _collect_subs, @@ -44,32 +37,27 @@ chain_skip_gate, inline_skip_gate, ) +from codeguard.hook_state import completed_event, record_completed_event, session_scope +from detect_lang import ensure_user_path, load_user_config +from gate_lib import ( # 兼容原有 Python 导入接口 + check_commit_safety, + format_safety_report, + gate_directive, + record_skip_event, + run_gate, + skip_gate_via_git_config, +) -# 保留旧导入面；实现分别属于纯语法与仓库观察适配器。 __all__ = [ "GUARDED_PATTERNS", "_analyze_segment", "_collect_subs", "_command_indirect", "_fallback_roots", "_filter_fallback_roots", "_flatten_substitutions", "_git_c_path", "_git_seg_parts", "_git_side_effect_sub", "_guarded_mode", "_is_git_repo", "_normalize_segment", "_repo_root_of", "_segment_is_git_side_effect", "chain_skip_gate", - "extract_command", "inline_skip_gate", "is_guarded", "main", "read_payload", - "resolve_project_roots", "staging_intent", + "check_commit_safety", "extract_command", "format_safety_report", "gate_directive", + "inline_skip_gate", "is_guarded", "main", "read_payload", "record_skip_event", + "resolve_project_roots", "run_gate", "skip_gate_via_git_config", "staging_intent", ] -from detect_lang import ( # # ensure_user_path/load_user_config 实际定义：scripts/paths.py、scripts/user_config.py - ensure_user_path, - load_user_config, -) -from gate_lib import ( - check_commit_safety, - format_safety_report, - gate_directive, - record_skip_event, - run_gate, - skip_gate_via_git_config, # 规范实现已上移 gate_lib（UPS 也要用） -) -from git_snapshot import SnapshotError - -# 拦截的 git 子命令（避免误拦 git status/diff/log 等只读命令） GUARDED_PATTERNS = ("git commit", "git push") @@ -84,24 +72,15 @@ def read_payload() -> dict: def extract_command(payload: dict) -> str: tool_input = payload.get("tool_input") or {} - cmd = tool_input.get("command") - return str(cmd) if cmd else "" - - + command = tool_input.get("command") + return str(command) if command else "" def main() -> int: - from codeguard.hook_state import ( - completed_event, - record_completed_event, - session_scope, - ) payload = read_payload() with session_scope(payload): if not payload.get("tool_use_id"): return _main(payload) - # 事件 ID 与命令、会话/worktree 一起绑定；只复用已结束的检查， - # 第一个副本尚在检查时，第二个副本宁可重跑，也不能直接放行。 key = "pre:" + str(payload["tool_use_id"]) + ":" + extract_command(payload) cached = completed_event(key) if cached is not None and cached["code"] == 2: @@ -112,133 +91,29 @@ def main() -> int: with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): code = _main(payload) out, err = stdout.getvalue(), stderr.getvalue() - # 只复用阻断结果，不缓存放行：无输出的 0 也可能来自显式豁免， - # 而未验证的 0 必须重跑。不能把事件去重缓存当作准确快照的替代。 - if code == 2: - record_completed_event(key, code, out, err) + # 只复用阻断结果，硬门放行和未验证永远重跑准确快照。 sys.stdout.write(out) sys.stderr.write(err) + if code == 2: + record_completed_event(key, code, out, err) return code def _main(payload: dict) -> int: ensure_user_path() - command = extract_command(payload) - mode = _guarded_mode(command) if command else None - if mode is None: - return 0 - - if os.environ.get("CODEGUARD_SKIP_GATE"): - record_skip_event("env") - return 0 - - cfg = load_user_config() - roots = resolve_project_roots(command) - # 静默跳过漏报修复：roots=[] 时扫描 cwd 一层子目录找 git 仓作兜底 - # （monorepo 模式：workspace 根目录下多仓时，从根目录直跑 git push 不 - # 经 cd 也能被覆盖）。找不到则向 #2 抛错，让 Agent 知晓绕过发生。 - fallback_note: str | None = None - if not roots: - fallback = _fallback_roots(Path(os.getcwd())) - if fallback: - roots = fallback - fallback_note = ( - f"未检测到显式 cd 仓；兜底扫描 cwd 一层子目录找到 {len(fallback)} 个 git 仓" - ) - else: - # 输出明确错误而非静默 exit 0：避免 Agent/用户误以为门禁执行过。 - print( - f"[codeguard] 未检测到任何 git 仓：cwd={os.getcwd()} 且命令链中无 cd <仓>。" - f"请在 git push 前先 cd <仓路径>，或在仓库根设置 git config codeguard.skipGate true。" - f"（已用兜底扫描 cwd 一层子目录，无 git 仓。）", - file=sys.stderr, - ) - 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], detail=fallback_note) - if any(skip_gate_via_git_config(r) for r in roots): - record_skip_event("skipGate", roots[0]) - return 0 - # 单次内联豁免：`git -c codeguard.skipGate=true …` 是显式意图（不落配置、 - # 无残留），放行并留审计明细——比仓库级 config 更不易"忘记恢复"。 - if inline_skip_gate(command): - record_skip_event("inline-skipGate", roots[0]) - print(json.dumps({"hookSpecificOutput": { - "hookEventName": "PreToolUse", "additionalContext": - "codeguard: 已通过内联豁免 `-c codeguard.skipGate` 放行本次提交" - "（已审计记录，Stop 摘要可见）。"}} - )) - return 0 - # 链内设置豁免：`git config codeguard.skipGate true && … && git config --unset …` - # 是文档推荐的自清理写法；此前整链被拦连 set 段都没跑成（首用必败）。 - if chain_skip_gate(command): - record_skip_event("chain-skipGate", roots[0]) + result = evaluate_git_command( + extract_command(payload), + cwd=Path(os.getcwd()), + load_config=load_user_config, + bypass_env=bool(os.environ.get("CODEGUARD_SKIP_GATE")), + ) + for context in result.contexts: print(json.dumps({"hookSpecificOutput": { - "hookEventName": "PreToolUse", "additionalContext": - "codeguard: 已通过链式豁免（set→提交→unset 同链）放行本次提交" - "（已审计记录，Stop 摘要可见）。"}} - )) - return 0 - - # 按命令链预测实际提交面（纯 commit → 仅 staged；add -A/-a → 三路）， - # 并入 git add 显式路径（add 尚未执行、暂存区还是旧的）。 - pending_commit = "commit" in _collect_subs(command) - - # 每个被操作的仓库独立跑：linter 门禁 + 提交内容安全检查 - # （commit 查暂存区；push 查未推送提交的 diff，防已提交未发现的坏文件） - reports = [] - for project_root in roots: - try: - lanes, extra = staging_intent(command, project_root=project_root) - except SnapshotError as exc: - print(json.dumps({"hookSpecificOutput": { - "hookEventName": "PreToolUse", "additionalContext": - f"codeguard: git UNVERIFIED：{project_root} 无法准确预测暂存面：{exc}；" - "请完成明确的暂存操作后单独提交，并复核检查结果。", - }}, ensure_ascii=False)) - continue - # 保留删除路径用于影响分析；单文件 linter 自行过滤不存在的文件。 - root_extra = list(extra) - failures, _skipped = run_gate( - project_root, cfg, mode=mode, lanes=lanes, extra=root_extra, - exact=True, pending_commit=pending_commit, - ) - if failures: - # 版本自标识由 gate_directive 首行综述之后的第二行承担—— - # 此处不再前置横幅：stderr 首行必须是综述（三个契约测试锁定）。 - 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: - print(json.dumps({"hookSpecificOutput": { - "hookEventName": "PreToolUse", "additionalContext": - "codeguard: 存在未验证项，不能宣称全部通过：" + "；".join(unknown), - }}, ensure_ascii=False)) - violations = check_commit_safety(project_root, mode, lanes=lanes, extra=root_extra, - pending_commit=pending_commit) - if violations: - reports.append(format_safety_report(violations) + ( - "\n\n**给 AI 的强制指令**：**密钥/凭据类**（.env、*.pem、id_* 等）" - "立即 git rm --cached 并提醒用户轮换密钥，不能只删了事；" - "**非密钥类**（依赖/产物目录、仓根 db/log 等）**先与用户确认**" - "是否为有意入库的第一方代码或合法 fixture——确认属误入库才执行" - "git rm --cached + .gitignore，确认后重新执行本次 git 命令。" - )) - - if not reports: - return 0 - # 硬拦截：stderr 作为工具结果反馈给 AI，AI 修复后重试本命令 - print(("\n\n" + "=" * 20 + " 下一个仓库 " + "=" * 20 + "\n\n").join(reports), file=sys.stderr) - return 2 + "hookEventName": "PreToolUse", "additionalContext": context, + }}, ensure_ascii=False)) + if result.stderr: + print(result.stderr, file=sys.stderr) + return result.exit_code if __name__ == "__main__": @@ -246,6 +121,6 @@ def _main(payload: dict) -> int: sys.exit(main()) except SystemExit: raise - except Exception as exc: # noqa: BLE001 — 内部错误 fail-open：traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open，不阻断宿主工具 print(f"[codeguard] 内部错误已忽略（fail-open）: {exc!r}", file=sys.stderr) sys.exit(0) diff --git a/hooks/stop_summary.py b/hooks/stop_summary.py index 8ab5ced..72b1263 100755 --- a/hooks/stop_summary.py +++ b/hooks/stop_summary.py @@ -18,56 +18,23 @@ sys.path.insert(0, str(PLUGIN_ROOT / "hooks")) sys.path.insert(0, str(PLUGIN_ROOT / "scripts")) +from codeguard import session_application + # 现行状态目录在 ~/.codeguard（CODEGUARD_HOME 可覆盖）；插件根的旧文件 # 升版本即丢、双副本各存一半——load_state 优先新路径，回退读旧文件一次。 LEGACY_STATE_FILE = PLUGIN_ROOT / ".session_state.json" def load_state() -> dict: - from codeguard.hook_state import state_path - from codeguard.storage import read_json - try: - return read_json(state_path(LEGACY_STATE_FILE)) - except OSError: - return {} + return session_application.load_state(LEGACY_STATE_FILE) def summarize(state: dict) -> str: - lang_state = {k: v for k, v in state.items() if not k.startswith("_")} - meta = {k: v for k, v in state.items() if k.startswith("_")} - if not lang_state and not meta: - return "[codeguard] 本次会话无 linter 记录（可能没有改代码文件）" - lines = ["[codeguard] 本次会话代码 lint 检查总结", ""] - for lang, info in lang_state.items(): - if not isinstance(info, dict): - continue - total = info.get("total", 0) - passed = info.get("passed", 0) - failed = info.get("failed", 0) - auto_fixed = info.get("auto_fixed", 0) - lines.append(f"- {lang}: 共 {total} 次检查，{passed} 通过，{failed} 失败，{auto_fixed} 自动修复成功") - skip = meta.get("_skip") or {} - if skip.get("count"): - kinds = "、".join(f"{k}×{v}" for k, v in (skip.get("kinds") or {}).items()) - lines.append(f"- ⚠️ 本会话绕过门禁 {skip['count']} 次（{kinds}）——确认是用户明确要求，并尽快 git config --unset codeguard.skipGate") - return "\n".join(lines) + return session_application.summarize(state) def _skipgate_left_enabled() -> bool: - """会话结束时提醒：仓库级豁免还开着（忘记 unset = 门禁永久静默失效）。""" - try: - import subprocess - - from gate_lib import skip_gate_via_git_config - proc = subprocess.run( - ["git", "rev-parse", "--git-dir"], - cwd=Path.cwd(), capture_output=True, check=False, text=True, timeout=10, - ) - if proc.returncode != 0: - return False - return skip_gate_via_git_config(Path.cwd()) - except Exception: # noqa: BLE001 — 提醒类检查，失败绝不影响 Stop - return False + return session_application.skipgate_left_enabled(Path.cwd()) def main(payload: dict | None = None) -> int: @@ -78,13 +45,8 @@ def main(payload: dict | None = None) -> int: def _main(payload: dict | None = None) -> int: - from codeguard.hook_state import state_path - from codeguard.storage import take_json - state = take_json(state_path(LEGACY_STATE_FILE)) - print(summarize(state)) - if _skipgate_left_enabled(): - print("[codeguard] ⚠️ git config codeguard.skipGate 仍为 true——该仓门禁处于豁免状态；" - "若绕过已完成，请执行 git config --unset codeguard.skipGate 恢复") + for line in session_application.consume_summary(LEGACY_STATE_FILE, Path.cwd()): + print(line) return 0 diff --git a/hooks/user_prompt_validator.py b/hooks/user_prompt_validator.py index b8d1ee8..a9d0396 100755 --- a/hooks/user_prompt_validator.py +++ b/hooks/user_prompt_validator.py @@ -1,13 +1,5 @@ #!/usr/bin/env python3 -"""UserPromptSubmit 钩子：用户说"提交/push/部署"时先跑 linter 门禁。 - -失败时【不阻断 prompt】（exit 2 会把用户消息弹回去，AI 收不到任何指令， -用户被卡死在输入框）——而是 exit 0 + stdout JSON additionalContext： -AI 收到「门禁未通过 + 修复指令」后自动修复、修完自己重新提交，形成闭环。 - -硬保证由 PreToolUse 钩子（pre_tool_git_guard.py）承担：AI 真去执行 -git commit/push 时被拦下，工具级 stderr 会作为结果反馈给 AI 继续修。 -""" +"""UserPromptSubmit 宿主适配器：软提醒注入，不阻断用户消息。""" from __future__ import annotations import contextlib @@ -16,24 +8,20 @@ import sys from pathlib import Path -PLUGIN_ROOT = Path(__file__).resolve().parents[1] # hooks/ 的上级 = 插件根（不依赖宿主环境变量） +PLUGIN_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PLUGIN_ROOT / "scripts")) sys.path.insert(0, str(PLUGIN_ROOT / "hooks")) -# 触发此钩子的关键词（中英） -# 英文按词边界匹配（防止 pushed/deployment 误命中），中文子串即可。 -# 2026-09-22 实测回归：`我刚才 pushed 了`、`关于 deployment 策略的讨论` -# 必须静默（子串匹配曾把它们当提交意图跑全仓 lint）。 -import re as _re - -from detect_lang import ( # # ensure_user_path/load_user_config 实际定义：scripts/paths.py、scripts/user_config.py +from codeguard import prompt_application, prompt_policy +from codeguard.hook_state import session_scope +from detect_lang import ( # 兼容原有 Python 导入接口 LANG_COMMANDS, detect_languages, ensure_user_path, find_project_root, load_user_config, ) -from gate_lib import ( +from gate_lib import ( # 兼容原有 Python 导入接口 check_commit_safety, format_safety_report, gate_directive, @@ -43,100 +31,38 @@ summarize_failures, ) -TRIGGER_PATTERNS = [ - "commit", "push", "deploy", "提交", "发布", "部署", +__all__ = [ + "LANG_COMMANDS", "QUESTION_MARKERS", "TRIGGER_PATTERNS", "check_commit_safety", + "detect_languages", "ensure_user_path", "find_project_root", "format_safety_report", + "gate_directive", "is_trigger", "load_user_config", "main", "notify", "read_payload", + "read_user_text", "record_skip_event", "run_gate", "skip_gate_via_git_config", + "summarize_failures", ] -_TRIGGER_RE = _re.compile( - r"\b(?:commit|push|deploy)\b|提交|发布|部署", - _re.IGNORECASE, -) -# 行动意图短语：触发词出现且（句首祈使 或 命中行动短语）才进门禁。 -_ACTION_INTENT_RE = _re.compile( - r"请帮我|帮我|请|马上|立刻|现在|下一步|继续|please|now|next|proceed|go ahead|" - r"commit this|push this|deploy this", - _re.IGNORECASE, -) + +TRIGGER_PATTERNS = prompt_policy.TRIGGER_PATTERNS +QUESTION_MARKERS = prompt_policy.QUESTION_MARKERS +is_trigger = prompt_policy.is_trigger +_detect_languages_in_text = prompt_policy.detect_languages_in_text +_non_git_note = prompt_application.non_git_note def notify(title: str, message: str) -> None: - """macOS 系统通知（非 darwin 静默；仅失败时打扰，通过靠注入文本确认）""" + """macOS 系统通知；其他宿主仅依赖对话上下文。""" if sys.platform != "darwin": return import subprocess - safe_t = title.replace('"', "'") - safe_m = message.replace('"', "'")[:200] + safe_title = title.replace('"', "'") + safe_message = message.replace('"', "'")[:200] with contextlib.suppress(OSError): subprocess.Popen( ["osascript", "-e", - f'display notification "{safe_m}" with title "{safe_t}" sound name "Pop"'], + f'display notification "{safe_message}" with title "{safe_title}" sound name "Pop"'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) -# 疑问句特征：在询问功能/做法，不是真的要提交（实测误触发： -# 「git 提交和推送，是不能把 .venv 排除掉么」被当成提交意图跑了全仓门禁） -QUESTION_MARKERS = ("？", "?", "么", "吗", "如何", "怎么", "有没有", "是不是", "什么是", "哪些") - - -def is_trigger(user_text: str) -> bool: - """触发条件：(1) 命中触发词（英文按词边界）(2) 非疑问句 (3) 行动意图或句首祈使。 - - (3) 缺一不可：`我刚才 pushed 了` 有触发词但无行动意图 → 静默； - `commit this now` 句首祈使 → 触发；`请帮我 commit` 行动短语 → 触发。 - """ - if not user_text: - return False - # 问句抑制按**句子粒度**：`这个方案 OK 吗？帮我提交` 问句与祈使混排时， - # 整条文本级的问号判断会把祈使句一起静默（软门缺失，硬门兜底，但漏提醒 - # 不符合"软门多提醒不算错"的既定原则）。 - # 切分时用捕获组保留定界符并附回句尾——"Should I commit this?" 的问号 - # 在句尾，若把定界符当纯分隔符吃掉，句内标记消失会误触发（CI 实测）。 - parts = _re.split(r"([。！!？?\n\r]+)", user_text) - sentences = [] - for i in range(0, len(parts) - 1, 2): - seg = (parts[i] + parts[i + 1]).strip() - if seg: - sentences.append(seg) - tail = parts[-1].strip() if len(parts) % 2 == 1 else "" - if tail: - sentences.append(tail) - trigger_sentences = [seg for seg in sentences if _TRIGGER_RE.search(seg)] - if not trigger_sentences: - return False - 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 = non_question[0].lstrip() - leading = stripped[: m.start()].strip().lower() - if leading in ("", "git", "to", "the", "a", "an", "帮我", "请", "请帮我"): - return True - return bool(_ACTION_INTENT_RE.search(user_text)) - - -def _detect_languages_in_text(user_text: str, language_ids: list[str]) -> list[str]: - """从用户消息中提取提到的语言子集（2.1/2.2/2.3）。 - - 与 `languages.json` 的 `id` 字段做大小写无关匹配；没有命中返回空列表， - 由调用方回退到 `detect_languages()` 全量探测（保持原有能力）。 - """ - if not user_text or not language_ids: - return [] - text = user_text.lower() - hits = [] - for lang_id in language_ids: - # 词边界避免 go 命中 gone/going、c 命中 cmake 等子串误命中。 - if _re.search(rf"\b{_re.escape(lang_id.lower())}\b", text): - hits.append(lang_id) - return hits - - def read_payload() -> dict: - """读一次 stdin payload（text 与 session_id 都从中取，stdin 只能读一次）。""" + """stdin 只读取一次，兼容 user_prompt 与 prompt 字段。""" if sys.stdin.isatty(): return {} try: @@ -151,126 +77,32 @@ def read_user_text() -> str: return str(payload.get("user_prompt") or payload.get("prompt") or "") -def _non_git_note() -> str: - return ( - "codeguard：当前目录不是 git 仓库，提交门禁已跳过" - "（门禁只对 git 仓生效；非 git 目录曾被回退成工作区根全仓扫描——" - "一个临时目录即可把上百个无关仓的存量 lint 变成永久红）。" - ) - - def main() -> int: - from codeguard.hook_state import session_scope payload = read_payload() with session_scope(payload): return _main(payload) def _main(payload: dict) -> int: - from codeguard.fingerprint import check_identity - from codeguard.hook_state import completed_event, record_completed_event - user_text = str(payload.get("user_prompt") or payload.get("prompt") or "") if not is_trigger(user_text): return 0 - - session_id = payload.get("session_id") - - ensure_user_path(from_login_shell=True) # GUI 宿主 PATH 不含用户级工具目录 - # 逃生门：设置此环境变量后跳过提交门禁（用于确实需要绕过的场景） - if os.environ.get("CODEGUARD_SKIP_GATE"): - record_skip_event("env") - return 0 - - root = find_project_root(os.getcwd()) - if root is None or not (root / ".git").exists(): - # 非 git 目录绝不回退成"扫这个目录"——曾经的 find_project_root or cwd - # 让工作区根成了全仓扫描对象（永久红且不可合法修复）。显式说明+跳过。 - print(json.dumps({ - "hookSpecificOutput": { - "hookEventName": "UserPromptSubmit", - "additionalContext": _non_git_note(), - } - }, ensure_ascii=False)) - return 0 - project_root = root - if skip_gate_via_git_config(project_root): - # 软门禁与硬门禁共用同一条仓库级豁免：此前 UPS 不认 skipGate， - # 出现过"硬门放行、软门仍喊禁止提交"的自相矛盾。 - record_skip_event("skipGate", project_root) - return 0 - - cfg = load_user_config() - # 2.2: 消息里提到了具体语言时只跑子集；没提到则回退全量探测（2.3）。 - detected = detect_languages(project_root) - subset = _detect_languages_in_text(user_text, detected) - def current_event_key(): - if not session_id: - return None - identity = check_identity(project_root, cfg, - {lang: LANG_COMMANDS.get(lang, {}) for lang in (subset or detected)}) - return f"ups:{session_id}:{user_text}:{identity}" if identity is not None else None - event_key = current_event_key() - if event_key and completed_event(event_key) is not None: - return 0 - # 推送意图走 push 面（含未推送提交），提交意图走 commit 面。 - # 文件面：软门禁**不传 lanes** → 三路宽口径（staged+未暂存+未跟踪）—— - # 此刻还没有待执行命令可预测，按"工作树有待提交改动就提醒"注入；硬门禁 - # （PreToolUse）知道真实命令形态，按 staging_intent 收窄到实际提交面。 - # 两者可以分歧：软门多提醒不算错（注入非阻断），硬门少拦才是底线。 - mode = "push" if _re.search(r"\bpush\b|推送", user_text, _re.IGNORECASE) else "commit" - failures, skipped = run_gate(project_root, cfg, languages=(subset or None), mode=mode) - # 提交内容安全检查：.venv/node_modules/.env/密钥等不应入库 - violations = check_commit_safety(project_root, mode) - - if failures or violations: - # 软引导：prompt 正常送达 AI，且注入修复指令，AI 自动修复后重新提交。 - # 注意：可能"只有安全违规、零 lint 失败"——此前 notify 直接取 - # failures[0][0] 越界，fail-open 吞成空输出（实测复现），两路都要兜。 - _first_issue = failures[0][1].splitlines()[0][:120] if failures and failures[0][1] else "详见对话" - _headline = ( - f"{failures[0][0]}: {_first_issue}" if failures - else (f"提交安全: {violations[0][0]}" if violations else "提交门禁发现违规") - ) - notify(summarize_failures(failures), _headline) - parts = [gate_directive(failures)] - if violations: - parts.append(format_safety_report(violations)) - parts.append( - "**给 AI 的强制指令**：**密钥/凭据类**必须 git rm --cached 并提醒用户" - "轮换密钥；**非密钥类**先与用户确认是否为有意入库的第一方代码或合法" - " fixture，确认误入库再 git rm --cached + 补 .gitignore；确认并修复后" - "重新执行提交。" - ) - print(json.dumps({ - "hookSpecificOutput": { - "hookEventName": "UserPromptSubmit", - "additionalContext": "\n\n".join(parts) - } - }, ensure_ascii=False)) - if event_key and not skipped and event_key == current_event_key(): - record_completed_event(event_key, 0, "", "") - return 0 - - # 通过：注入明确的成功确认（无此行用户会以为门禁根本没跑）； - # 即使宿主把它渲染成提示条，文案也是明确的「✅ 通过」语义 - skipped_langs = {s.split()[0] for s in skipped} - checked = sorted(set(detect_languages(project_root)) - skipped_langs) - skipped_note = f"；跳过 {len(skipped)} 项（{'；'.join(skipped)}）" if skipped else "" - unknown = [s for s in skipped if "本次改动未涉及" not in s and " SKIPPED:" not in s - and "markdown 风格告警" not in s] - print(json.dumps({ - "hookSpecificOutput": { + ensure_user_path(from_login_shell=True) + result = prompt_application.evaluate_prompt( + user_text, + project_root=find_project_root(os.getcwd()), + session_id=payload.get("session_id"), + load_config=load_user_config, + bypass_env=bool(os.environ.get("CODEGUARD_SKIP_GATE")), + ) + if result.notification: + notify(*result.notification) + if result.additional_context is not None: + print(json.dumps({"hookSpecificOutput": { "hookEventName": "UserPromptSubmit", - "additionalContext": ( - ("codeguard ⚠️ 检查范围存在未验证项，不能宣称全部通过" if unknown - else f"codeguard ✅ 提交门禁通过：已检查 {len(checked)} 个语言生态 + 暂存区安全") - + f"（{', '.join(checked) or '无'}）{skipped_note}。" - ) - } - }, ensure_ascii=False)) - if event_key and not skipped and event_key == current_event_key(): - record_completed_event(event_key, 0, "", "") + "additionalContext": result.additional_context, + }}, ensure_ascii=False)) + result.acknowledge() return 0 @@ -279,6 +111,6 @@ def current_event_key(): sys.exit(main()) except SystemExit: raise - except Exception as exc: # noqa: BLE001 — 内部错误 fail-open：traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open，不阻断用户消息 print(f"[codeguard] 内部错误已忽略（fail-open）: {exc!r}", file=sys.stderr) sys.exit(0) diff --git a/kimi.plugin.json b/kimi.plugin.json index 457f7ff..a9ad206 100644 --- a/kimi.plugin.json +++ b/kimi.plugin.json @@ -1,6 +1,6 @@ { "name": "codeguard", - "version": "0.13.0", + "version": "0.14.0", "description": "Evidence-backed code checks and Git content gates for AI assistants, with Maven/Gradle module impact analysis. Save hooks provide feedback; unverified checks are explicit.", "author": { "name": "Full Stack Skills / PartMe.AI" diff --git a/openspec/changes/refactor-codeguard-architecture/tasks.md b/openspec/changes/refactor-codeguard-architecture/tasks.md index 712cfcb..4eb3c4d 100644 --- a/openspec/changes/refactor-codeguard-architecture/tasks.md +++ b/openspec/changes/refactor-codeguard-architecture/tasks.md @@ -26,7 +26,7 @@ - [x] 4.1 统一原子状态读改写、审计及已完成事件边界；6 进程 210 次统计/审计、并发消费 240 次计数、真实会话/worktree 与未知重试测试通过。硬门禁重复拒绝仍为 2，放行不缓存；旧无 ID 状态明确不保证隔离。Windows 锁实现未实机验收；内容/配置缓存继续 4.2。 - [x] 4.2 分离 fingerprint/cache，绑定 worktree/index 内容/upstream/文件内容/配置/命令；真实等长同 mtime 修改、linked worktree、未知恢复、配置更新和硬门禁不复用软缓存等 16 项通过。输入变化/预算超限不写缓存；软观察不是原子快照，外部动态输入与大型仓库性能边界已记录。 -- [ ] 4.3 CLI/MCP/五个 hooks 成为协议适配入口，保留用户可见行为；语言检查已归 `language_check`、请求策略已归 `check_application`，CLI/MCP 协议专项通过。五个 hook 仍需逐个收敛，不能据此勾选整体完成。 +- [x] 4.3 CLI/MCP/五个 hooks 成为协议适配入口，保留用户可见行为；语言检查归 `language_check`、请求策略归 `check_application`，五类 Hook 分别委托 startup/session/prompt/git_guard/save 应用服务。独立导入、真实会话/Git、宿主 JSON 与 fail-open 回归通过；旧 Python 导入继续兼容。仍不把 CLI/MCP/Hook 的本地契约测试称为三宿主安装验收。 ## 5. 质量与交付 @@ -34,5 +34,5 @@ - [x] 5.2 四个历史 skipped 已全部恢复；Dockerfile CLI 接入统一报告解析与扫描应用，14 项真实进程边界测试通过；基线比较按诊断内容和次数核验并要求基线实际 FAIL，三项误豁免用例先红后绿。完整 433 项单测零跳过、142 项回归零失败。 - [x] 5.3 新增当前架构 Mermaid 与扩展指南；报告版本读取插件 manifest，双语 README 指向当前证据，旧设计标明历史范围；双语标题/链接/版本及本地链接存在性测试通过。发布版本仍需在 5.5 检查。 - [ ] 5.4 全量单测/真实 hook 回归/ruff/语言 schema/vendor 离线与在线/本 change strict 和逐条规格追溯通过。 -- [ ] 5.5 按仓规升级版本、生成市场元数据、授权的 PR/CI/不可变 tag/Release 闭环；宿主安装运行独立记录，未授权不得改宿主。 +- [ ] 5.5 按仓规升级版本、生成市场元数据、授权的 PR/CI/不可变 tag/Release 闭环；v0.13.0 阶段版本已完成插件/市场双仓合并、PR/main CI、tag/Release，但后续工作树改动仍未发布。宿主安装运行独立记录，未授权不得改宿主。 - [ ] 5.6 完成全目标审计，代码/规格/证据一致后同步并归档 change，才可将整体优化标记完成。 diff --git a/openspec/changes/refactor-codeguard-architecture/verification.md b/openspec/changes/refactor-codeguard-architecture/verification.md index 779c1f5..eae139e 100644 --- a/openspec/changes/refactor-codeguard-architecture/verification.md +++ b/openspec/changes/refactor-codeguard-architecture/verification.md @@ -147,3 +147,18 @@ 5. 通用执行器仅处理文本检查命令；Git blob 的二进制协议必须保留字节，不得为了“一致”强制 UTF-8 解码。 6. 第一期不解决任意命令进程树管理、完整 shell 解释、沙箱及所有动态 Java 构建；这些边界不能在文档中消失。 7. 只审阅了 pre-commit 官方的通用语言协议与 Python 适配器作边界参考，未声称移植或达到同等成熟度。 + +## 第十三批实际验证：v0.13.0 阶段发布与会话 Hook 应用层 + +- 上一批源码作为阶段版本 v0.13.0 经插件 PR #45、市场 PR #1 合并到两个 `main`；插件 PR 与 main 的 `skills-check` 均通过。不可变 `v0.13.0` 注释 tag 解引用到插件 main 合并提交 `fb95bbad6a6748e081918b79e58d7d1bb85a7d40`，GitHub Release 已发布；市场 Codex/ZCode/Kimi 清单为 0.13.0，生成器只读校验通过。三宿主已安装副本和运行现场仍未验证，不以发布代替宿主验收。 +- 本批新改动先有 RED：SessionStart/Stop 应用服务不存在，3+3 项新契约测试失败。`startup_application` 接管语言发现、配置盘点、探活、双副本/版本提醒报告；`session_application` 接管原子状态消费、总结与残留 skipGate 观察。两个 Hook 保留原路径、旧 Python 导入与会话去重，仅做宿主配置、通知、stdout 和 fail-open。 +- 真实 Git 和会话测试确认 Stop 只消费当前作用域一次、残留 skipGate 单列提醒；独立解释器证明两个应用服务导入不加载 Hook 或宿主 SDK。修改后的临时 CodeGraph 快照同步了 8 个文件，调用链显示 `build_startup_report` 与 `consume_summary` 均由对应 Hook 调用，未在源仓初始化图谱。 +- 本地本批全量 **441 unittest 全部通过、0 skipped**；`tests/run_all.py` **142/0/0**；ruff、架构依赖门禁、语言 schema 57 项/11 规则、vendor 离线与在线、本 change strict、diff whitespace 均通过。PostToolUse、UserPromptSubmit、PreToolUse 仍持有应用编排，4.3 和整体目标继续进行；本批代码尚未 bump、提交、推送或发布。 + +## 第十四批实际验证：五类 Hook 协议入口收敛 + +- UserPromptSubmit 的意图/语言策略与软门禁编排分别进入 `prompt_policy`/`prompt_application`；PreToolUse 的 Git 命令判定进入 `git_guard_application`，并显式传入调用者 cwd；PostToolUse 的保存检查、修复复检与反馈进入 `save_application`。SessionStart/Stop 沿用上一批应用服务。五个 Hook 保留原路径、协议输出、通知、fail-open 与兼容导入。 +- RED 后 GREEN：五个应用服务的独立导入与行为测试；软提示/保存的完成确认只在宿主输出后发生。PreToolUse 故障注入曾证明拦截事件在 stdout 写入前被记为完成，现改为输出后再记；宿主输出异常时不会错误缓存已交付拦截。既有 `run_all.py` 的异常注入改打实际 `prompt_application.run_gate` 所有者，并保持 fail-open 断言。 +- 临时 CodeGraph 快照重新同步了本批改动并追踪三个新应用服务的 Hook 调用点；源仓未初始化索引。`scripts/check_architecture.py` 声明并约束新内核模块，隔离导入测试阻止其反向加载宿主 Hook/MCP。 +- 使用本机已有、带 MCP SDK 的 `/opt/anaconda3/bin/python3` 3.13.5 完整执行 **454 unittest，全部通过、0 skipped**；`tests/run_all.py` **142/0/0**。Homebrew Python 3.14 缺 MCP SDK 时同一套测试仅执行 451 项、跳过 3 项，故不将该运行作为完整证明。ruff、架构门禁、语言 schema 57 项/11 规则、vendor 离线与在线、本 change strict、diff whitespace 通过；技能 lock 和受管技能未改。 +- 本地入口边界已满足 task 4.3。Stop 会先消费会话状态再输出，若 stdout 故障可能丢失该次总结；三宿主安装现场、Windows、联网漏洞库及所有语言真实工具链仍待独立验收。本批源码在本记录时仍未 bump/提交/推送/发布，不把 v0.13.0 的远端证据套用到新改动。 diff --git a/scripts/check_architecture.py b/scripts/check_architecture.py index 7ffd83f..2277b9a 100644 --- a/scripts/check_architecture.py +++ b/scripts/check_architecture.py @@ -32,7 +32,7 @@ "codeguard.planning": {"__future__", "collections", "pathlib", "scope", "codeguard.models"}, "codeguard.git_syntax": {"__future__", "pathlib", "re", "shlex"}, "codeguard.git_staging": {"__future__", "os", "re", "shlex", "pathlib", "git_snapshot", "codeguard.git_syntax"}, - "codeguard.git_context": {"__future__", "os", "pathlib", "re", "shlex", "scope", + "codeguard.git_context": {"__future__", "pathlib", "re", "shlex", "scope", "codeguard.execution", "codeguard.git_syntax", "codeguard.git_staging"}, "codeguard.path_policy": {"__future__", "fnmatch", "pathlib"}, "codeguard.storage": {"__future__", "contextlib", "json", "os", "tempfile", "time", @@ -57,6 +57,26 @@ "codeguard.check_application": {"__future__", "pathlib", "scope", "codeguard.config", "codeguard.discovery", "codeguard.java_analysis", "codeguard.language_check", "codeguard.registry"}, + "codeguard.session_application": {"__future__", "pathlib", "codeguard.execution", + "codeguard.hook_state", "codeguard.repository_policy", + "codeguard.storage"}, + "codeguard.startup_application": {"__future__", "re", "dataclasses", "pathlib", + "codeguard.config", "codeguard.discovery", + "codeguard.registry", "codeguard.toolchain"}, + "codeguard.prompt_policy": {"__future__", "re"}, + "codeguard.prompt_application": {"__future__", "collections", "dataclasses", "pathlib", + "codeguard.discovery", "codeguard.fingerprint", "codeguard.gate", + "codeguard.hook_state", "codeguard.prompt_policy", + "codeguard.registry", "codeguard.reporting", + "codeguard.repository_policy"}, + "codeguard.git_guard_application": {"__future__", "collections", "dataclasses", "pathlib", + "git_snapshot", "codeguard.gate", "codeguard.git_context", + "codeguard.git_syntax", "codeguard.hook_state", + "codeguard.reporting", "codeguard.repository_policy"}, + "codeguard.save_application": {"__future__", "hashlib", "tempfile", "collections", + "dataclasses", "pathlib", "scope", "codeguard.hook_state", + "codeguard.discovery", "codeguard.execution", "codeguard.fingerprint", + "codeguard.planning", "codeguard.registry", "codeguard.verdict"}, "codeguard.java_build": {"__future__", "re", "xml", "pathlib"}, "codeguard.java_impact": {"__future__", "collections", "pathlib"}, "codeguard.java_planning": {"__future__", "collections"}, diff --git a/scripts/codeguard/git_context.py b/scripts/codeguard/git_context.py index 9d734b4..0093489 100644 --- a/scripts/codeguard/git_context.py +++ b/scripts/codeguard/git_context.py @@ -1,7 +1,6 @@ """将静态 Git 意图绑定到本地仓库；负责文件/进程观察，不产生宿主输出。""" from __future__ import annotations -import os import shlex from pathlib import Path @@ -30,7 +29,7 @@ def _is_git_repo(p: Path) -> bool: return execute(["git", "-C", str(p), "rev-parse", "--git-dir"], Path.cwd(), 10).returncode == 0 -def resolve_project_roots(command: str) -> list[Path]: +def resolve_project_roots(command: str, cwd: Path | None = None) -> list[Path]: """顺序扫描命令链，收集每个 git commit/push 段各自的仓库边界。 链式发布（cd plugins && git commit && cd minimax && git push）操作 @@ -47,7 +46,8 @@ def resolve_project_roots(command: str) -> list[Path]: command = _flatten_substitutions(command) roots: list[Path] = [] last_cd: Path | None = None - current_dir = Path.cwd().resolve() + caller_dir = (cwd or Path.cwd()).resolve() + current_dir = caller_dir for seg in re.split(r"&&|\|\||;|\n", command): seg = seg.strip() if not seg: @@ -67,7 +67,7 @@ def resolve_project_roots(command: str) -> list[Path]: if c_path is not None: root = c_path if c_path.is_dir() and _is_git_repo(c_path) else None else: - root = last_cd or (Path(os.getcwd()) if _is_git_repo(Path(os.getcwd())) else None) + root = last_cd or (caller_dir if _is_git_repo(caller_dir) else None) if root: root = repository_root(root) if root not in roots: diff --git a/scripts/codeguard/git_guard_application.py b/scripts/codeguard/git_guard_application.py new file mode 100644 index 0000000..fdcc4ad --- /dev/null +++ b/scripts/codeguard/git_guard_application.py @@ -0,0 +1,128 @@ +"""PreToolUse Git 硬门禁应用：仓库解析、豁免、准确检查与结构化决策。""" +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +from git_snapshot import SnapshotError + +from .gate import run_gate +from .git_context import ( + _fallback_roots as fallback_roots, +) +from .git_context import ( + _filter_fallback_roots as filter_fallback_roots, +) +from .git_context import ( + _guarded_mode as guarded_mode, +) +from .git_context import ( + resolve_project_roots, + staging_intent, +) +from .git_syntax import _collect_subs as collect_subs +from .git_syntax import chain_skip_gate, inline_skip_gate +from .hook_state import record_skip_event +from .reporting import format_safety_report, gate_directive +from .repository_policy import check_commit_safety, skip_gate_via_git_config + + +@dataclass(frozen=True) +class GitGuardResult: + """宿主按 contexts 输出 additionalContext，按 stderr/exit_code 决定拦截。""" + + exit_code: int = 0 + contexts: tuple[str, ...] = () + stderr: str = "" + + +def evaluate_git_command(command: str, *, cwd: Path, load_config: Callable[[], dict], + bypass_env: bool = False) -> GitGuardResult: + """评估一条工具命令；不修改 index，不直接输出宿主协议。""" + mode = guarded_mode(command) if command else None + if mode is None: + return GitGuardResult() + if bypass_env: + record_skip_event("env") + return GitGuardResult() + + cfg = load_config() + roots = resolve_project_roots(command, cwd=cwd) + fallback_note: str | None = None + if not roots: + fallback = fallback_roots(cwd) + if fallback: + roots = fallback + fallback_note = f"未检测到显式 cd 仓；兜底扫描 cwd 一层子目录找到 {len(fallback)} 个 git 仓" + else: + return GitGuardResult( + 2, + stderr=(f"[codeguard] 未检测到任何 git 仓：cwd={cwd} 且命令链中无 cd <仓>。" + "请在 git push 前先 cd <仓路径>，或在仓库根设置 git config codeguard.skipGate true。" + "（已用兜底扫描 cwd 一层子目录，无 git 仓。）"), + ) + 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 GitGuardResult() + record_skip_event("monorepo-fallback", roots[0], detail=fallback_note) + if any(skip_gate_via_git_config(root) for root in roots): + record_skip_event("skipGate", roots[0]) + return GitGuardResult() + if inline_skip_gate(command): + record_skip_event("inline-skipGate", roots[0]) + context = ( + "codeguard: 已通过内联豁免 `-c codeguard.skipGate` 放行本次提交" + "（已审计记录，Stop 摘要可见）。" + ) + return GitGuardResult(contexts=(context,)) + if chain_skip_gate(command): + record_skip_event("chain-skipGate", roots[0]) + context = ( + "codeguard: 已通过链式豁免（set→提交→unset 同链）放行本次提交" + "（已审计记录，Stop 摘要可见）。" + ) + return GitGuardResult(contexts=(context,)) + + pending_commit = "commit" in collect_subs(command) + contexts: list[str] = [] + reports: list[str] = [] + for project_root in roots: + try: + lanes, extra = staging_intent(command, project_root=project_root) + except SnapshotError as exc: + contexts.append( + f"codeguard: git UNVERIFIED：{project_root} 无法准确预测暂存面：{exc}；" + "请完成明确的暂存操作后单独提交，并复核检查结果。" + ) + continue + root_extra = list(extra) + failures, skipped = run_gate( + project_root, cfg, mode=mode, lanes=lanes, extra=root_extra, + exact=True, pending_commit=pending_commit, + ) + if failures: + reports.append(gate_directive(failures, project_root=project_root)) + unknown = [item for item in skipped if "本次改动未涉及" not in item and " SKIPPED:" not in item + and "markdown 风格告警" not in item] + if unknown: + contexts.append("codeguard: 存在未验证项，不能宣称全部通过：" + "；".join(unknown)) + violations = check_commit_safety(project_root, mode, lanes=lanes, extra=root_extra, + pending_commit=pending_commit) + if violations: + reports.append(format_safety_report(violations) + ( + "\n\n**给 AI 的强制指令**：**密钥/凭据类**（.env、*.pem、id_* 等）" + "立即 git rm --cached 并提醒用户轮换密钥，不能只删了事；" + "**非密钥类**（依赖/产物目录、仓根 db/log 等）**先与用户确认**" + "是否为有意入库的第一方代码或合法 fixture——确认属误入库才执行" + "git rm --cached + .gitignore，确认后重新执行本次 git 命令。" + )) + + if not reports: + return GitGuardResult(contexts=tuple(contexts)) + detail = ("\n\n" + "=" * 20 + " 下一个仓库 " + "=" * 20 + "\n\n").join(reports) + return GitGuardResult(2, tuple(contexts), detail) diff --git a/scripts/codeguard/prompt_application.py b/scripts/codeguard/prompt_application.py new file mode 100644 index 0000000..7c9e918 --- /dev/null +++ b/scripts/codeguard/prompt_application.py @@ -0,0 +1,106 @@ +"""UserPromptSubmit 软门禁应用：返回上下文与通知，不写宿主协议。""" +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path + +from .discovery import detect_languages +from .fingerprint import check_identity +from .gate import run_gate +from .hook_state import completed_event, record_completed_event, record_skip_event +from .prompt_policy import detect_languages_in_text, intent_mode, is_trigger +from .registry import LANG_COMMANDS +from .reporting import format_safety_report, gate_directive, summarize_failures +from .repository_policy import check_commit_safety, skip_gate_via_git_config + + +@dataclass(frozen=True) +class PromptResult: + """软门禁结果：由 Hook 适配器编码为 additionalContext。""" + + additional_context: str | None = None + notification: tuple[str, str] | None = None + _on_delivered: Callable[[], None] | None = field(default=None, repr=False, compare=False) + + def acknowledge(self) -> None: + """宿主成功输出后才登记完成，输出失败时保留重试机会。""" + if self._on_delivered is not None: + self._on_delivered() + + +def non_git_note() -> str: + return ( + "codeguard：当前目录不是 git 仓库，提交门禁已跳过" + "（门禁只对 git 仓生效；非 git 目录曾被回退成工作区根全仓扫描——" + "一个临时目录即可把上百个无关仓的存量 lint 变成永久红）。" + ) + + +def evaluate_prompt(user_text: str, *, project_root: Path | None, + session_id: object, load_config: Callable[[], dict], + bypass_env: bool = False) -> PromptResult: + """评估提交意图，保留软提醒、逃生门、内容身份去重与宽范围观察。""" + if not is_trigger(user_text): + return PromptResult() + if bypass_env: + record_skip_event("env") + return PromptResult() + if project_root is None or not (project_root / ".git").exists(): + return PromptResult(non_git_note()) + if skip_gate_via_git_config(project_root): + record_skip_event("skipGate", project_root) + return PromptResult() + + cfg = load_config() + detected = detect_languages(project_root) + subset = detect_languages_in_text(user_text, detected) + + def current_event_key() -> str | None: + if not session_id: + return None + identity = check_identity(project_root, cfg, + {lang: LANG_COMMANDS.get(lang, {}) for lang in (subset or detected)}) + return f"ups:{session_id}:{user_text}:{identity}" if identity is not None else None + + event_key = current_event_key() + if event_key and completed_event(event_key) is not None: + return PromptResult() + + # 软门没有待执行 git 命令，故保持三路宽口径；硬门另用预测暂存面。 + mode = intent_mode(user_text) + failures, skipped = run_gate(project_root, cfg, languages=(subset or None), mode=mode) + violations = check_commit_safety(project_root, mode) + + def on_delivered() -> None: + if event_key == current_event_key(): + record_completed_event(event_key, 0, "", "") + + completion = on_delivered if event_key and not skipped else None + + if failures or violations: + first_issue = failures[0][1].splitlines()[0][:120] if failures and failures[0][1] else "详见对话" + headline = (f"{failures[0][0]}: {first_issue}" if failures + else f"提交安全: {violations[0][0]}") + parts = [gate_directive(failures)] + if violations: + parts.append(format_safety_report(violations)) + parts.append( + "**给 AI 的强制指令**：**密钥/凭据类**必须 git rm --cached 并提醒用户" + "轮换密钥；**非密钥类**先与用户确认是否为有意入库的第一方代码或合法" + " fixture，确认误入库再 git rm --cached + 补 .gitignore；确认并修复后" + "重新执行提交。" + ) + return PromptResult("\n\n".join(parts), (summarize_failures(failures), headline), completion) + + skipped_langs = {item.split()[0] for item in skipped} + checked = sorted(set(detect_languages(project_root)) - skipped_langs) + skipped_note = f"；跳过 {len(skipped)} 项（{'；'.join(skipped)}）" if skipped else "" + unknown = [item for item in skipped if "本次改动未涉及" not in item and " SKIPPED:" not in item + and "markdown 风格告警" not in item] + context = ( + ("codeguard ⚠️ 检查范围存在未验证项，不能宣称全部通过" if unknown + else f"codeguard ✅ 提交门禁通过：已检查 {len(checked)} 个语言生态 + 暂存区安全") + + f"（{', '.join(checked) or '无'}）{skipped_note}。" + ) + return PromptResult(context, _on_delivered=completion) diff --git a/scripts/codeguard/prompt_policy.py b/scripts/codeguard/prompt_policy.py new file mode 100644 index 0000000..d926ae0 --- /dev/null +++ b/scripts/codeguard/prompt_policy.py @@ -0,0 +1,56 @@ +"""用户提交意图与语言子集的纯策略，不读取宿主事件或执行门禁。""" +from __future__ import annotations + +import re + +TRIGGER_PATTERNS = ["commit", "push", "deploy", "提交", "发布", "部署"] +_TRIGGER_RE = re.compile(r"\b(?:commit|push|deploy)\b|提交|发布|部署", re.IGNORECASE) +_ACTION_INTENT_RE = re.compile( + r"请帮我|帮我|请|马上|立刻|现在|下一步|继续|please|now|next|proceed|go ahead|" + r"commit this|push this|deploy this", + re.IGNORECASE, +) +QUESTION_MARKERS = ("？", "?", "么", "吗", "如何", "怎么", "有没有", "是不是", "什么是", "哪些") + + +def is_trigger(user_text: str) -> bool: + """仅行动句触发软门禁；疑问句、陈述既往动作和词内子串不触发。""" + if not user_text: + return False + parts = re.split(r"([。！!？?\n\r]+)", user_text) + sentences = [] + for i in range(0, len(parts) - 1, 2): + segment = (parts[i] + parts[i + 1]).strip() + if segment: + sentences.append(segment) + tail = parts[-1].strip() if len(parts) % 2 == 1 else "" + if tail: + sentences.append(tail) + trigger_sentences = [segment for segment in sentences if _TRIGGER_RE.search(segment)] + if not trigger_sentences: + return False + non_question = [segment for segment in trigger_sentences + if not any(marker in segment for marker in QUESTION_MARKERS)] + if not non_question: + return False + match = _TRIGGER_RE.search(non_question[0]) + if not match: + return False + leading = non_question[0].lstrip()[:match.start()].strip().lower() + if leading in ("", "git", "to", "the", "a", "an", "帮我", "请", "请帮我"): + return True + return bool(_ACTION_INTENT_RE.search(user_text)) + + +def detect_languages_in_text(user_text: str, language_ids: list[str]) -> list[str]: + """用户明确提到语言时只选择对应注册项；无命中由调用方回退全量。""" + if not user_text or not language_ids: + return [] + text = user_text.lower() + return [lang_id for lang_id in language_ids + if re.search(rf"\b{re.escape(lang_id.lower())}\b", text)] + + +def intent_mode(user_text: str) -> str: + """推送意图包含未推送提交；其余提交/发布意图按 commit 面观察。""" + return "push" if re.search(r"\bpush\b|推送", user_text, re.IGNORECASE) else "commit" diff --git a/scripts/codeguard/save_application.py b/scripts/codeguard/save_application.py new file mode 100644 index 0000000..5b228f6 --- /dev/null +++ b/scripts/codeguard/save_application.py @@ -0,0 +1,201 @@ +"""PostToolUse 保存检查应用：单文件范围、修复复检与有证据的反馈。""" +from __future__ import annotations + +import hashlib +import tempfile +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path + +from scope import is_build_artifact + +from . import hook_state +from .discovery import detect_language, project_uses_linter +from .execution import execute +from .fingerprint import check_identity +from .hook_state import codeguard_home +from .planning import scoped_plan +from .registry import LANG_COMMANDS +from .verdict import UNVERIFIED, lint_verdict + +PLUGIN_ROOT = Path(__file__).resolve().parents[2] + + +@dataclass(frozen=True) +class SaveResult: + """宿主输出 additionalContext 后确认统计；通知由宿主适配器执行。""" + + additional_context: str | None = None + system_message: str | None = None + notification: tuple[str, str, str] | None = None + _on_delivered: Callable[[], None] | None = field(default=None, repr=False, compare=False) + + def acknowledge(self) -> None: + if self._on_delivered is not None: + self._on_delivered() + + +def state_path() -> Path: + """兼容旧无 ID 状态，带会话 ID 时绝不认领无归属记录。""" + return hook_state.state_path(PLUGIN_ROOT / ".session_state.json") + + +def dedup_path() -> Path: + return codeguard_home() / "file_dedup.json" + + +def bump_state(lang: str, passed: bool, auto_fixed: bool = False) -> None: + hook_state.bump_state(state_path(), lang, passed, auto_fixed) + + +def notification_allowed(lang: str, cooldown_s: int = 60) -> bool: + return hook_state.notification_allowed(state_path(), lang, cooldown_s) + + +def should_suppress_duplicate(file_path: str, identity: str | None, + store: Path | None = None) -> bool: + return hook_state.completed_file_check(file_path, store or dedup_path(), identity) + + +def is_ejs_template(file_path: str) -> bool: + """EJS 标记不是 HTML，htmlhint 会对模板语法产生必然误报。""" + try: + return "<%" in Path(file_path).read_text(errors="ignore")[:8192] + except OSError: + return False + + +def run(cmd: list[str], cwd: Path, timeout: int = 300) -> tuple[int, str, str]: + """检查器与 formatter 使用共享进程边界，保留原 tuple 兼容语义。""" + return execute(cmd, cwd, timeout).as_tuple() + + +def should_skip(file_path: str, languages: list[str]) -> tuple[bool, str]: + if not file_path or is_build_artifact(file_path): + return True, "" + lang = detect_language(file_path) + if not lang: + return True, "" + if languages and lang not in languages: + return True, lang + return False, lang + + +def _porcelain(project_root: Path) -> set[str]: + """修复影响提示的尽力观察，不作为是否通过的证据。""" + outcome = execute(["git", "status", "--porcelain"], project_root, 10) + if outcome.failure or outcome.returncode != 0: + return set() + return {line[3:].strip() for line in outcome.stdout.splitlines() if len(line) > 3} + + +def _failure_context(file_path: str, lang: str, cmd_def: dict, stdout: str, + stderr: str) -> tuple[str, str]: + raw = (stdout or stderr or "").strip() + problem_lines = [line.strip() for line in raw.splitlines() if line.strip()] + problems = "\n".join(problem_lines[:5])[:500] or "（linter 未输出具体问题）" + if len(problem_lines) > 5 or len(raw) > 500: + try: + key = hashlib.sha1(str(file_path).encode()).hexdigest()[:12] + full_path = Path(tempfile.gettempdir()) / f"codeguard-post-{lang}-{key}.log" + full_path.write_text(raw, encoding="utf-8") + problems += f"\n…（截断，共 {len(problem_lines)} 行；完整输出: {full_path}）" + except OSError: + problems += f"\n…（截断，共 {len(problem_lines)} 行）" + fix_cmd = " ".join(cmd_def.get("format") or []) or "按上述问题逐条修复" + context = ( + f"codeguard ⚠️ [{lang}] {Path(file_path).name} 检查未通过\n" + f"发现的问题（节选）:\n{problems}\n" + "怎么修:\n" + f" 1. 可先运行 `{fix_cmd}` 自动修复格式类问题\n" + " 2. 手动修复上述具体问题后，重新保存该文件，codeguard 会自动复检" + ) + return context, problem_lines[0][:120] if problem_lines else "" + + +def evaluate_save(file_path: str, project_root: Path, cfg: dict, + *, store: Path | None = None) -> SaveResult: + """冻结 save 计划并执行一次检查；必要时只修复并复检同一文件。""" + skipped, lang = should_skip(file_path, cfg.get("enabled_languages", [])) + if skipped or lang == "markdown": + return SaveResult() + cmd_def = LANG_COMMANDS.get(lang) + if not cmd_def: + return SaveResult() + if not cmd_def.get("append_files", True): + return SaveResult(f"codeguard: {lang} 是项目级检查；本次保存未验证，" + "请使用 codeguard check 或提交门禁。不会自动格式化整个项目。") + if not project_uses_linter(cmd_def, project_root): + return SaveResult() + if lang == "html" and is_ejs_template(file_path): + return SaveResult() + lint_cmd = cmd_def.get("lint") or [] + if not lint_cmd: + return SaveResult() + + timeout = cfg.get("lint_timeout_seconds", 120) + lint_plan = scoped_plan(lint_cmd, project_root, scope="save", files=[str(file_path)]) + lint_argv = list(lint_plan.commands[0].argv) + store = store or dedup_path() + + def identity() -> str | None: + return check_identity(project_root, cfg, {lang: cmd_def}, extra_files=[file_path]) + + checked_identity = identity() + if checked_identity is not None and should_suppress_duplicate(file_path, checked_identity, store): + return SaveResult() + rc, stdout, stderr = run(lint_argv, cwd=lint_plan.cwd, timeout=timeout) + verdict, reason = lint_verdict(rc, lint_cmd, stdout + stderr) + if verdict == UNVERIFIED: + return SaveResult(f"codeguard: UNVERIFIED [{lang}] {reason} (exit {rc})；" + "没有代码结论，不自动修复。") + + auto_fixed = False + touched_others: list[str] = [] + if rc != 0 and cfg.get("auto_fix_on_save", True): + fmt_cmd = cmd_def.get("format") + if fmt_cmd: + before = _porcelain(project_root) + fix_plan = scoped_plan(fmt_cmd, project_root, scope="save", files=[str(file_path)]) + fix_rc, _, _ = run(list(fix_plan.commands[0].argv), cwd=fix_plan.cwd, timeout=timeout + 60) + if fix_rc == 0: + auto_fixed = True + after = _porcelain(project_root) + touched_others = sorted(name for name in (after - before) + if not name.rstrip("/").endswith(Path(file_path).name)) + checked_identity = identity() + rc, stdout, stderr = run(lint_argv, cwd=lint_plan.cwd, timeout=timeout) + + verdict, reason = lint_verdict(rc, lint_cmd, stdout + stderr) + if verdict == UNVERIFIED: + return SaveResult(f"codeguard: UNVERIFIED [{lang}] 修复后复检未完成：{reason}；" + "formatter 可能已修改文件，请重新读取。没有通过/失败结论。") + + passed = rc == 0 + + def on_delivered() -> None: + if checked_identity is not None and checked_identity == identity(): + hook_state.record_file_check(file_path, store, checked_identity) + bump_state(lang, passed=passed, auto_fixed=auto_fixed) + + if passed: + note = "" + if auto_fixed: + note = f"（已自动运行 {' '.join(cmd_def.get('format') or [])} 修复）" + if touched_others: + shown = ", ".join(touched_others[:10]) + more = f" 等 {len(touched_others)} 个" if len(touched_others) > 10 else "" + note += (f"。⚠️ format 命令还改动了本文件之外的文件{more}：{shown}" + "——这些文件你内存里的版本已过期，必须先重新读取再继续操作") + else: + note += "（仅本文件被改动，请重新读取）" + return SaveResult( + f"codeguard: ✅ {lang} lint passed for {Path(file_path).name}{note}", + f"codeguard: ✅ {lang} lint passed", + _on_delivered=on_delivered, + ) + + context, first_problem = _failure_context(file_path, lang, cmd_def, stdout, stderr) + alert = f"{lang} lint FAILED: {Path(file_path).name}" + return SaveResult(context, f"codeguard: ⚠️ {alert}", + (lang, alert, first_problem), on_delivered) diff --git a/scripts/codeguard/session_application.py b/scripts/codeguard/session_application.py new file mode 100644 index 0000000..479008e --- /dev/null +++ b/scripts/codeguard/session_application.py @@ -0,0 +1,60 @@ +"""会话结束应用服务：消费本会话统计，返回不依赖宿主的展示行。""" +from __future__ import annotations + +from pathlib import Path + +from .execution import execute +from .hook_state import state_path +from .repository_policy import skip_gate_via_git_config +from .storage import read_json, take_json + + +def load_state(legacy_state_file: Path) -> dict: + """兼容只读访问旧状态路径；读取不消费记录。""" + try: + return read_json(state_path(legacy_state_file)) + except OSError: + return {} + + +def summarize(state: dict) -> str: + """将本次会话统计格式化为宿主无关的文本。""" + lang_state = {k: v for k, v in state.items() if not k.startswith("_")} + meta = {k: v for k, v in state.items() if k.startswith("_")} + if not lang_state and not meta: + return "[codeguard] 本次会话无 linter 记录（可能没有改代码文件）" + lines = ["[codeguard] 本次会话代码 lint 检查总结", ""] + for lang, info in lang_state.items(): + if not isinstance(info, dict): + continue + total = info.get("total", 0) + passed = info.get("passed", 0) + failed = info.get("failed", 0) + auto_fixed = info.get("auto_fixed", 0) + lines.append(f"- {lang}: 共 {total} 次检查，{passed} 通过，{failed} 失败，{auto_fixed} 自动修复成功") + skip = meta.get("_skip") or {} + if skip.get("count"): + kinds = "、".join(f"{k}×{v}" for k, v in (skip.get("kinds") or {}).items()) + lines.append(f"- ⚠️ 本会话绕过门禁 {skip['count']} 次（{kinds}）——确认是用户明确要求，并尽快 git config --unset codeguard.skipGate") + return "\n".join(lines) + + +def skipgate_left_enabled(project_root: Path) -> bool: + """提醒类观察；Git 或配置读取失败不改变 Stop 的退出协议。""" + try: + probe = execute(["git", "rev-parse", "--git-dir"], project_root, 10) + if probe.failure or probe.returncode != 0: + return False + return skip_gate_via_git_config(project_root) + except Exception: # noqa: BLE001 — 可选提醒不得阻断会话结束 + return False + + +def consume_summary(legacy_state_file: Path, project_root: Path) -> tuple[str, ...]: + """原子消费当前作用域状态，并返回需要由宿主输出的消息。""" + state = take_json(state_path(legacy_state_file)) + lines = [summarize(state)] + if skipgate_left_enabled(project_root): + lines.append("[codeguard] ⚠️ git config codeguard.skipGate 仍为 true——该仓门禁处于豁免状态；" + "若绕过已完成，请执行 git config --unset codeguard.skipGate 恢复") + return tuple(lines) diff --git a/scripts/codeguard/startup_application.py b/scripts/codeguard/startup_application.py new file mode 100644 index 0000000..60c5813 --- /dev/null +++ b/scripts/codeguard/startup_application.py @@ -0,0 +1,130 @@ +"""SessionStart 应用服务：项目盘点、工具可用性与上下文内容。""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path + +from .config import load_user_config +from .discovery import detect_languages +from .registry import LANG_COMMANDS, REGISTRY +from .toolchain import ToolchainProbe + + +@dataclass(frozen=True) +class StartupReport: + """宿主适配器负责呈现 text，并在必要时发送 notification。""" + + text: str + notification: tuple[str, str] | None = None + + +def detect_linter_config(project_root: Path) -> dict: + """从语言注册表盘点项目已有的 linter 配置。""" + found = {} + for lang_id, lang_def in REGISTRY.items(): + files = lang_def.get("linter_config_files") or [] + if not files: + continue + present = [f for f in files if (project_root / f).exists()] + if present: + found[lang_id] = present + return found + + +def _semver_tuple(version: str) -> tuple[int, ...]: + match = re.match(r"(\d+(?:\.\d+)*)", version.strip()) + if not match: + return (0,) + return tuple(int(part) for part in match.group(1).split(".")) + + +def version_backlog_note(current: str, cache_root: Path) -> str | None: + """离线比较当前版本与 ZCode 缓存中可用的版本目录。""" + newer: list[str] = [] + installed: list[str] = [] + if not cache_root.is_dir(): + return None + try: + for vendor_dir in sorted(cache_root.glob("*/codeguard/*")): + if not vendor_dir.is_dir(): + continue + version = vendor_dir.name + installed.append(version) + if _semver_tuple(version) > _semver_tuple(current): + newer.append(version) + except OSError: + return None + if not newer: + return None + newest = max(newer, key=_semver_tuple) + return ( + f"- ⚠️ **版本积压**：当前生效 codeguard `v{current}`，本机 cache 已有更新版本 " + f"`v{newest}`（共 {len(installed)} 份: {', '.join(sorted(installed))}）——" + "钩子行为以**生效副本**为准，修复可能已存在却没跑到；请更新插件并清理旧 " + "cache（`~/.zcode/cli/plugins/cache/*/codeguard/`），同时核对远端是否还有" + "更新版本（本检测离线，看不到仓库最新版）。" + ) + + +def build_startup_report(project_root: Path, *, plugin_version: str = "", + cache_root: Path | None = None, + enabled_plugin_orgs: tuple[str, ...] = ()) -> StartupReport: + """盘点语言和工具；不打印、不通知，也不读取宿主插件注册文件。""" + languages = detect_languages(project_root) + if not languages: + return StartupReport("") + + linter_cfg = detect_linter_config(project_root) + cfg = load_user_config() + enabled = cfg.get("enabled_languages", []) + if enabled and enabled != ["auto"]: + languages = [lang for lang in languages if lang in enabled] + + probe = ToolchainProbe(project_root) + ready, missing = [], [] + for lang in languages: + cmd_def = LANG_COMMANDS.get(lang) + if not cmd_def: + continue + ok, reason = probe.probe(cmd_def) + if ok: + ready.append(lang) + else: + hint = cmd_def.get("install_hint") or "见 docs/LANGUAGES.md" + missing.append((lang, reason, hint)) + + lines = ["# codeguard 项目记忆", "", f"- 项目根: `{project_root}`", + f"- 检测到语言: {', '.join(languages)}"] + notification = None + if missing: + lines.extend(["", "**⚠️ 以下语言的检查未生效，这些语言的文件【不会】被检查（提交门禁也会跳过它们）：**"]) + for lang, reason, hint in missing: + lines.append(f" - {lang}（{reason}）→ 修复: `{hint}`") + lines.append(" 请主动提示用户安装；安装后新写的文件才会被检查。") + notification = ( + "codeguard 部分检查未生效", + f"{len(missing)} 个语言未生效: " + ", ".join(lang for lang, _, _ in missing) + "；详情见对话", + ) + if ready: + lines.append(f"- 检查已生效的语言: {', '.join(ready)}") + if linter_cfg: + lines.append(f"- 已有的 linter 配置: {linter_cfg}") + else: + lines.append("- 未配置任何 linter（建议运行 /init 接入 codeguard）") + + if len(enabled_plugin_orgs) > 1: + lines.append( + f"- ⚠️ 检测到 codeguard **双副本同时启用**（{', '.join(sorted(enabled_plugin_orgs))}）：" + "每个钩子事件会执行两遍，报告与统计可能翻倍或分裂——建议只保留一个来源" + ) + if plugin_version and cache_root is not None: + note = version_backlog_note(plugin_version, cache_root) + if note: + lines.append(note) + + lines.append("- AI 写完代码会被 PostToolUse 钩子自动 lint，告警会出现在这里，按告警里的「怎么修」处理") + lines.append("- 用户要求「提交/push」时，UserPromptSubmit 钩子会再次确认所有 linter 通过，未通过会拦截提交") + lines.append("") + lines.append("**重要**：写代码前先阅读对应语言的规范（见 skills/codeguard-{language}/SKILL.md）") + return StartupReport("\n".join(lines), notification) diff --git a/tests/run_all.py b/tests/run_all.py index 5ee1150..ce27db4 100755 --- a/tests/run_all.py +++ b/tests/run_all.py @@ -570,10 +570,11 @@ def counting(root, cfg, langs, **kw): "sys.argv=['hook']\n" "sys.stdin=io.StringIO(json.dumps({'user_prompt':'提交代码'}))\n" "sys.path.insert(0,'hooks')\n" - "import gate_lib\n" + "sys.path.insert(0,'scripts')\n" + "from codeguard import prompt_application\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 配置 + "prompt_application.run_gate=_boom\n" + "prompt_application.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, diff --git a/tests/test_architecture.py b/tests/test_architecture.py index fa0d437..6de99af 100644 --- a/tests/test_architecture.py +++ b/tests/test_architecture.py @@ -107,6 +107,51 @@ def test_report_version_matches_installed_plugin_manifest(self): self.assertEqual(0, result.returncode, result.stderr) self.assertEqual(expected, result.stdout.strip()) + def test_session_application_import_does_not_load_hook_adapter(self): + result = subprocess.run([sys.executable, "-I", "-c", + ("import sys;sys.path.insert(0,sys.argv[1]);" + "import codeguard.session_application;" + "assert not any(m in sys.modules for m in " + "('stop_summary','gate_lib','mcp'))"), + str(ROOT / "scripts")], capture_output=True, text=True, timeout=10, check=False) + self.assertEqual(0, result.returncode, result.stderr) + + def test_startup_application_import_does_not_load_hook_adapter(self): + result = subprocess.run([sys.executable, "-I", "-c", + ("import sys;sys.path.insert(0,sys.argv[1]);" + "import codeguard.startup_application;" + "assert not any(m in sys.modules for m in " + "('env_check','gate_lib','mcp'))"), + str(ROOT / "scripts")], capture_output=True, text=True, timeout=10, check=False) + self.assertEqual(0, result.returncode, result.stderr) + + def test_prompt_application_import_does_not_load_hook_adapter(self): + result = subprocess.run([sys.executable, "-I", "-c", + ("import sys;sys.path.insert(0,sys.argv[1]);" + "import codeguard.prompt_application;" + "assert not any(m in sys.modules for m in " + "('user_prompt_validator','gate_lib','mcp'))"), + str(ROOT / "scripts")], capture_output=True, text=True, timeout=10, check=False) + self.assertEqual(0, result.returncode, result.stderr) + + def test_git_guard_application_import_does_not_load_hook_adapter(self): + result = subprocess.run([sys.executable, "-I", "-c", + ("import sys;sys.path.insert(0,sys.argv[1]);" + "import codeguard.git_guard_application;" + "assert not any(m in sys.modules for m in " + "('pre_tool_git_guard','gate_lib','mcp'))"), + str(ROOT / "scripts")], capture_output=True, text=True, timeout=10, check=False) + self.assertEqual(0, result.returncode, result.stderr) + + def test_save_application_import_does_not_load_hook_adapter(self): + result = subprocess.run([sys.executable, "-I", "-c", + ("import sys;sys.path.insert(0,sys.argv[1]);" + "import codeguard.save_application;" + "assert not any(m in sys.modules for m in " + "('post_tool_lint','gate_lib','mcp'))"), + str(ROOT / "scripts")], capture_output=True, text=True, timeout=10, check=False) + self.assertEqual(0, result.returncode, result.stderr) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_check_plans.py b/tests/test_check_plans.py index 2e4e9c1..3c51ea4 100644 --- a/tests/test_check_plans.py +++ b/tests/test_check_plans.py @@ -18,7 +18,7 @@ import gate_lib import post_tool_lint import run_per_language -from codeguard import execution +from codeguard import execution, save_application from detect_lang import LANG_COMMANDS @@ -157,7 +157,7 @@ def test_save_fix_and_recheck_stay_on_original_file(self): patch.object(post_tool_lint, "read_payload", return_value={"file_path": str(target)}), \ patch.object(post_tool_lint, "find_project_root", return_value=self.root), \ patch.object(post_tool_lint, "load_user_config", return_value={"auto_fix_on_save": True}), \ - patch.object(post_tool_lint, "should_suppress_duplicate", return_value=False), \ + patch.object(save_application, "should_suppress_duplicate", return_value=False), \ patch.dict(os.environ, {"CODEGUARD_HOME": str(self.root / "state")}), \ contextlib.redirect_stdout(output): self.assertEqual(0, post_tool_lint.main()) diff --git a/tests/test_git_boundaries.py b/tests/test_git_boundaries.py index e045db5..8fe460c 100644 --- a/tests/test_git_boundaries.py +++ b/tests/test_git_boundaries.py @@ -64,6 +64,16 @@ def test_repository_binding_and_indirection_work_without_hooks_on_sys_path(self) left, right, script) self.assertEqual([[str(left), str(right)], "push", ["staged", "unstaged"], False], actual) + def test_repository_resolution_uses_explicit_cwd_not_process_cwd(self): + with tempfile.TemporaryDirectory(prefix="cg-explicit-cwd-") as tmp: + repo = Path(tmp).resolve() + subprocess.run(["git", "init", "-q", str(repo)], check=True, capture_output=True) + actual = self.isolated( + "from codeguard.git_context import resolve_project_roots;import json;" + "print(json.dumps([str(p) for p in resolve_project_roots('git push',cwd=__import__('pathlib').Path(sys.argv[2]))]))", + repo) + self.assertEqual([str(repo)], actual) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_git_guard_application.py b/tests/test_git_guard_application.py new file mode 100644 index 0000000..1ff753e --- /dev/null +++ b/tests/test_git_guard_application.py @@ -0,0 +1,74 @@ +"""PreToolUse 的 Git 决策返回结构化结果，不直接写宿主 stdout/stderr。""" +from __future__ import annotations + +import contextlib +import io +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path[:0] = [str(ROOT / "scripts"), str(ROOT / "hooks")] + +import pre_tool_git_guard + + +class GitGuardApplicationTests(unittest.TestCase): + def test_no_git_repository_blocks_with_diagnostic_without_printing(self): + from codeguard import git_guard_application + + with tempfile.TemporaryDirectory(prefix="cg-guard-app-") as tmp: + output, errors = io.StringIO(), io.StringIO() + with patch.object(git_guard_application, "resolve_project_roots", return_value=[]), \ + patch.object(git_guard_application, "fallback_roots", return_value=[]), \ + contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors): + result = git_guard_application.evaluate_git_command( + "git push", cwd=Path(tmp), load_config=dict) + self.assertEqual(2, result.exit_code) + self.assertIn("未检测到任何 git 仓", result.stderr) + self.assertEqual((), result.contexts) + self.assertEqual("", output.getvalue() + errors.getvalue()) + + def test_inline_bypass_returns_context_and_audit_without_printing(self): + from codeguard import git_guard_application + + with tempfile.TemporaryDirectory(prefix="cg-guard-inline-") as tmp: + root = Path(tmp) + output = io.StringIO() + with patch.object(git_guard_application, "resolve_project_roots", return_value=[root]), \ + patch.object(git_guard_application, "skip_gate_via_git_config", return_value=False), \ + patch.object(git_guard_application, "inline_skip_gate", return_value=True), \ + patch.object(git_guard_application, "record_skip_event") as record, \ + contextlib.redirect_stdout(output): + result = git_guard_application.evaluate_git_command( + "git -c codeguard.skipGate=true commit -m ok", cwd=root, load_config=dict) + self.assertEqual(0, result.exit_code) + self.assertIn("内联豁免", result.contexts[0]) + record.assert_called_once() + self.assertEqual("", output.getvalue()) + + def test_blocked_event_is_not_marked_complete_before_host_output(self): + class BrokenOutput: + def write(self, _text): + raise OSError("host output unavailable") + + payload = {"tool_use_id": "event-1", "tool_input": {"command": "git commit -m x"}} + + def block(_payload): + print("blocked") + return 2 + + with patch.object(pre_tool_git_guard, "read_payload", return_value=payload), \ + patch.object(pre_tool_git_guard, "_main", side_effect=block), \ + patch.object(pre_tool_git_guard, "completed_event", return_value=None), \ + patch.object(pre_tool_git_guard, "record_completed_event") as record, \ + patch.object(sys, "stdout", BrokenOutput()), \ + self.assertRaises(OSError): + pre_tool_git_guard.main() + record.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_prompt_application.py b/tests/test_prompt_application.py new file mode 100644 index 0000000..5fcb0da --- /dev/null +++ b/tests/test_prompt_application.py @@ -0,0 +1,67 @@ +"""UserPromptSubmit 应用层返回软提醒，不直接占用 stdout 或宿主通知。""" +from __future__ import annotations + +import contextlib +import io +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + + +class PromptApplicationTests(unittest.TestCase): + def test_non_git_prompt_does_not_read_config_or_emit_host_output(self): + from codeguard.prompt_application import evaluate_prompt + + output = io.StringIO() + with contextlib.redirect_stdout(output): + result = evaluate_prompt("提交代码", project_root=None, session_id=None, + load_config=lambda: self.fail("非 Git 目录不应加载配置")) + self.assertEqual("", output.getvalue()) + self.assertIn("不是 git 仓库", result.additional_context) + self.assertIsNone(result.notification) + + def test_safety_only_violation_has_soft_context_and_no_blocking_exit(self): + from codeguard import prompt_application + + with tempfile.TemporaryDirectory(prefix="cg-prompt-app-") as tmp: + repo = Path(tmp) + subprocess.run(["git", "init", "-q", str(repo)], check=True, capture_output=True) + with patch.object(prompt_application, "skip_gate_via_git_config", return_value=False), \ + patch.object(prompt_application, "detect_languages", return_value=[]), \ + patch.object(prompt_application, "run_gate", return_value=([], [])), \ + patch.object(prompt_application, "check_commit_safety", + return_value=[(".env", "密钥", "git rm --cached .env")]): + result = prompt_application.evaluate_prompt( + "提交代码", project_root=repo, session_id=None, load_config=dict) + self.assertIn(".env", result.additional_context) + self.assertIn("密钥/凭据", result.additional_context) + self.assertIsNotNone(result.notification) + + def test_completed_event_is_recorded_only_after_host_delivery(self): + from codeguard import prompt_application + + with tempfile.TemporaryDirectory(prefix="cg-prompt-delivery-") as tmp: + repo = Path(tmp) + subprocess.run(["git", "init", "-q", str(repo)], check=True, capture_output=True) + with patch.object(prompt_application, "skip_gate_via_git_config", return_value=False), \ + patch.object(prompt_application, "detect_languages", return_value=["python"]), \ + patch.object(prompt_application, "check_identity", return_value="content-id"), \ + patch.object(prompt_application, "completed_event", return_value=None), \ + patch.object(prompt_application, "run_gate", return_value=([], [])), \ + patch.object(prompt_application, "check_commit_safety", return_value=[]), \ + patch.object(prompt_application, "record_completed_event") as record: + result = prompt_application.evaluate_prompt( + "提交代码", project_root=repo, session_id="session-1", load_config=dict) + record.assert_not_called() + result.acknowledge() + record.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_save_application.py b/tests/test_save_application.py new file mode 100644 index 0000000..3551f5f --- /dev/null +++ b/tests/test_save_application.py @@ -0,0 +1,77 @@ +"""PostToolUse 应用层负责保存检查，宿主适配器负责 JSON/通知。""" +from __future__ import annotations + +import contextlib +import io +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + + +class SaveApplicationTests(unittest.TestCase): + def test_project_only_checker_reports_unverified_without_running(self): + from codeguard import save_application + + with tempfile.TemporaryDirectory(prefix="cg-save-app-") as tmp: + root = Path(tmp) + file_path = root / "main.py" + file_path.write_text("print(1)\n", encoding="utf-8") + output = io.StringIO() + with patch.object(save_application, "detect_language", return_value="python"), \ + patch.dict(save_application.LANG_COMMANDS, + {"python": {"lint": ["python", "-V"], "append_files": False}}), \ + patch.object(save_application, "run") as run, \ + contextlib.redirect_stdout(output): + result = save_application.evaluate_save(str(file_path), root, {}) + self.assertEqual("", output.getvalue()) + self.assertIn("项目级检查", result.additional_context) + run.assert_not_called() + + def test_unverified_lint_never_auto_fixes_or_records_completion(self): + from codeguard import save_application + + with tempfile.TemporaryDirectory(prefix="cg-save-unknown-") as tmp: + root = Path(tmp) + file_path = root / "main.py" + file_path.write_text("print(1)\n", encoding="utf-8") + with patch.object(save_application, "detect_language", return_value="python"), \ + patch.dict(save_application.LANG_COMMANDS, + {"python": {"lint": ["ruff", "check", "{file}"], + "format": ["ruff", "format", "{file}"]}}), \ + patch.object(save_application, "project_uses_linter", return_value=True), \ + patch.object(save_application, "check_identity", return_value=None), \ + patch.object(save_application, "run", return_value=(2, "", "config invalid")) as run, \ + patch.object(save_application, "bump_state") as bump: + result = save_application.evaluate_save(str(file_path), root, {"auto_fix_on_save": True}) + result.acknowledge() + self.assertIn("UNVERIFIED", result.additional_context) + self.assertEqual(1, run.call_count) + bump.assert_not_called() + + def test_passed_check_records_statistics_only_after_delivery(self): + from codeguard import save_application + + with tempfile.TemporaryDirectory(prefix="cg-save-delivery-") as tmp: + root = Path(tmp) + file_path = root / "main.py" + file_path.write_text("print(1)\n", encoding="utf-8") + with patch.object(save_application, "detect_language", return_value="python"), \ + patch.dict(save_application.LANG_COMMANDS, + {"python": {"lint": ["ruff", "check", "{file}"]}}), \ + patch.object(save_application, "project_uses_linter", return_value=True), \ + patch.object(save_application, "check_identity", return_value=None), \ + patch.object(save_application, "run", return_value=(0, "", "")), \ + patch.object(save_application, "bump_state") as bump: + result = save_application.evaluate_save(str(file_path), root, {}) + bump.assert_not_called() + result.acknowledge() + bump.assert_called_once_with("python", passed=True, auto_fixed=False) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_session_application.py b/tests/test_session_application.py new file mode 100644 index 0000000..97da722 --- /dev/null +++ b/tests/test_session_application.py @@ -0,0 +1,56 @@ +"""Stop 应用服务只返回结果，宿主适配器负责 stdout 与 fail-open 协议。""" +from __future__ import annotations + +import contextlib +import io +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + + +class SessionApplicationTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory(prefix="cg-session-app-") + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.repo = self.root / "repo" + self.repo.mkdir() + subprocess.run(["git", "init", "-q", str(self.repo)], check=True, capture_output=True) + self.home = self.root / "state" + + def test_consume_summary_returns_lines_without_host_output_and_drains_once(self): + from codeguard.session_application import consume_summary + + self.home.mkdir() + (self.home / "session_state.json").write_text( + '{"python":{"total":2,"passed":1,"failed":1,"auto_fixed":0}}', encoding="utf-8") + output = io.StringIO() + with patch.dict(os.environ, {"CODEGUARD_HOME": str(self.home)}), \ + contextlib.redirect_stdout(output): + first = consume_summary(ROOT / ".session_state.json", self.repo) + second = consume_summary(ROOT / ".session_state.json", self.repo) + self.assertEqual("", output.getvalue()) + self.assertEqual(1, len(first)) + self.assertIn("共 2 次检查", first[0]) + self.assertIn("无 linter 记录", second[0]) + + def test_skipgate_reminder_is_a_separate_result_line(self): + from codeguard.session_application import consume_summary + + subprocess.run(["git", "config", "codeguard.skipGate", "true"], cwd=self.repo, + check=True, capture_output=True) + with patch.dict(os.environ, {"CODEGUARD_HOME": str(self.home)}): + lines = consume_summary(ROOT / ".session_state.json", self.repo) + self.assertEqual(2, len(lines)) + self.assertIn("仍为 true", lines[1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_startup_application.py b/tests/test_startup_application.py new file mode 100644 index 0000000..c3b5ca3 --- /dev/null +++ b/tests/test_startup_application.py @@ -0,0 +1,48 @@ +"""SessionStart 的发现、探活与报告不依赖宿主输出或通知。""" +from __future__ import annotations + +import contextlib +import io +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + + +class StartupApplicationTests(unittest.TestCase): + def test_no_detected_languages_is_silent(self): + from codeguard import startup_application + + with patch.object(startup_application, "detect_languages", return_value=[]): + report = startup_application.build_startup_report(Path.cwd()) + self.assertEqual("", report.text) + self.assertIsNone(report.notification) + + def test_missing_tool_returns_message_and_notification_without_printing(self): + from codeguard import startup_application + + with tempfile.TemporaryDirectory(prefix="cg-start-app-") as tmp: + project = Path(tmp) + output = io.StringIO() + with patch.object(startup_application, "detect_languages", return_value=["python"]), \ + patch.object(startup_application, "detect_linter_config", return_value={}), \ + patch.object(startup_application, "load_user_config", return_value={}), \ + patch.object(startup_application.ToolchainProbe, "probe", + return_value=(False, "检查器不可用")), \ + patch.dict(startup_application.LANG_COMMANDS, + {"python": {"install_hint": "安装 ruff"}}), \ + contextlib.redirect_stdout(output): + report = startup_application.build_startup_report( + project, plugin_version="0.13.0", enabled_plugin_orgs=("one", "two")) + self.assertEqual("", output.getvalue()) + self.assertIn("python（检查器不可用）", report.text) + self.assertIn("双副本同时启用", report.text) + self.assertEqual("codeguard 部分检查未生效", report.notification[0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_verdict_integrity.py b/tests/test_verdict_integrity.py index 872564f..0fda911 100644 --- a/tests/test_verdict_integrity.py +++ b/tests/test_verdict_integrity.py @@ -22,7 +22,7 @@ import pre_tool_git_guard import run_check import run_per_language -from codeguard import cve_scanners +from codeguard import cve_scanners, save_application from detect_lang import LANG_COMMANDS # auto_fix 用例需真实 ruff 执行 format/check——工具缺失时断言的修复路径无法覆盖。 @@ -154,7 +154,7 @@ def test_post_tool_error_does_not_run_formatter(self): with patch.dict(LANG_COMMANDS, {"python": commands}), \ patch.object(post_tool_lint, "read_payload", return_value={"file_path": str(file)}), \ patch.object(post_tool_lint, "find_project_root", return_value=self.root), \ - patch.object(post_tool_lint, "should_suppress_duplicate", return_value=False), \ + patch.object(save_application, "should_suppress_duplicate", return_value=False), \ patch.object(post_tool_lint, "mac_notify"), \ patch.dict(os.environ, {"CODEGUARD_HOME": str(self.root / "state")}), \ contextlib.redirect_stdout(io.StringIO()): @@ -168,14 +168,15 @@ def test_post_tool_recheck_error_is_json_unverified_not_failure(self): with patch.dict(LANG_COMMANDS, {"python": commands}), \ patch.object(post_tool_lint, "read_payload", return_value={"file_path": str(file)}), \ patch.object(post_tool_lint, "find_project_root", return_value=self.root), \ - patch.object(post_tool_lint, "should_suppress_duplicate", return_value=False), \ + patch.object(save_application, "should_suppress_duplicate", return_value=False), \ patch.object(post_tool_lint, "load_user_config", return_value={"auto_fix_on_save": True}), \ - patch.object(post_tool_lint, "run", side_effect=[(1, "F401", ""), (0, "", ""), (2, "", "config invalid")]), \ - patch.object(post_tool_lint, "bump_state") as bump, \ + patch.object(save_application, "run", side_effect=[(1, "F401", ""), (0, "", ""), (2, "", "config invalid")]) as run, \ + patch.object(save_application, "bump_state") as bump, \ patch.object(post_tool_lint, "mac_notify"), contextlib.redirect_stdout(out): post_tool_lint.main() payload = json.loads(out.getvalue()) self.assertIn("UNVERIFIED", payload["hookSpecificOutput"]["additionalContext"]) + self.assertEqual(3, run.call_count) bump.assert_not_called()