diff --git a/.github/workflows/skills-check.yml b/.github/workflows/skills-check.yml index 8822dc2..99c6303 100644 --- a/.github/workflows/skills-check.yml +++ b/.github/workflows/skills-check.yml @@ -15,6 +15,8 @@ jobs: - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: python-version: "3.12" + - name: Install requirements (mcp SDK so MCP server tests run, not skip) + run: python3 -m pip install --disable-pip-version-check -r requirements.txt - name: Verify vendored skills match the lockfile digests run: python3 scripts/vendor/skill_vendor.py check --offline - name: Verify lockfile pins still match upstream refs and content diff --git a/.gitignore b/.gitignore index fa5f80f..8083045 100644 --- a/.gitignore +++ b/.gitignore @@ -19,3 +19,5 @@ build/ dist/ htmlcov/ Thumbs.db +# codeguard 失败日志落盘目录 +out/ diff --git a/README.md b/README.md index 48c7116..a602a46 100644 --- a/README.md +++ b/README.md @@ -70,7 +70,7 @@ AI code that passes lint on first try | Current version | `0.6.7` | | ZCode manifest | `.zcode-plugin/plugin.json` | | Codex manifest | `.codex-plugin/plugin.json` | -| MCP server | Not published until the SDK-backed protocol implementation is ready; use the CLI and hooks | +| MCP server | Published: stdio server via the official SDK (`check_code_style` / `auto_fix` / `list_languages`); see Quick start | | Primary language | Python 3.10+ (hooks), YAML/JSON (config) | | License | Apache-2.0 | @@ -199,6 +199,25 @@ When this plugin is active, you do **not** need to do anything manually: - When the user says "commit", you will receive a final all-linters-must-pass gate check. - At session end, you will see a summary of which lints passed/failed. +### MCP server + +`run_check.py --mcp` starts a stdio MCP server (official `mcp` SDK; install +dependencies with `pip install -r requirements.txt`) exposing three tools: + +| Tool | Purpose | +|---|---| +| `check_code_style` | Run lint and return a per-language envelope with `stderr_path` and `log_path` | +| `auto_fix` | Run the formatter chain, then re-run lint and embed the check envelope | +| `list_languages` | List supported language ids and display names (no internal commands) | + +```bash +python3 scripts/run_check.py --mcp . +``` + +Failed runs write the full combined output to `/out/.codeguard-last.log` +(the same path appears in the tool envelope and the CLI summary line; suppress +with `--quiet`). + ## Configuration User config in `~/.zcode/settings.local.yaml` (ZCode) or equivalent for other hosts: diff --git a/README.zh-CN.md b/README.zh-CN.md index 637cd1b..5152f3d 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -70,7 +70,7 @@ AI 一次写出就过 lint 的代码 | 当前版本 | `0.6.7` | | ZCode manifest | `.zcode-plugin/plugin.json` | | Codex manifest | `.codex-plugin/plugin.json` | -| MCP 服务 | SDK 协议实现完成前不随清单发布;当前使用 CLI 与 Hooks | +| MCP 服务 | 已发布:官方 SDK stdio 服务(`check_code_style` / `auto_fix` / `list_languages`);见快速开始 | | 主要语言 | Python 3.10+(钩子)、YAML/JSON(配置) | | 协议 | Apache-2.0 | @@ -184,6 +184,24 @@ ln -s $PWD ~/.kimi/plugins/partme-codeguard-plugin - 用户说「commit」时,你会受到「所有 linter 必须通过」的最后门禁 - 会话结束时会看到本会话 lint 通过/失败的总结 +### MCP 服务器 + +`run_check.py --mcp` 启动 stdio MCP 服务(官方 `mcp` SDK;先用 +`pip install -r requirements.txt` 安装依赖),暴露三个工具: + +| 工具 | 用途 | +|---|---| +| `check_code_style` | 跑 lint,返回逐语言信封与 `stderr_path`、`log_path` | +| `auto_fix` | 先跑 formatter 链再复检 lint(嵌入 check 信封) | +| `list_languages` | 列出语言 id 与显示名(不含内部命令) | + +```bash +python3 scripts/run_check.py --mcp . +``` + +失败时完整输出落盘到 `<项目根>/out/.codeguard-last.log`(与工具信封及 CLI +摘要行里的路径一致;`--quiet` 可关闭)。 + ## 配置 `~/.zcode/settings.local.yaml`(ZCode)或各平台对应文件: diff --git a/hooks/env_check.py b/hooks/env_check.py index a15f1cf..da36836 100755 --- a/hooks/env_check.py +++ b/hooks/env_check.py @@ -7,7 +7,6 @@ 3. 输出一段 AGENTS.md 风格的提示,让 AI 知道自己在一个被 codeguard 管理的项目里 """ import os -import shutil import subprocess import sys from pathlib import Path @@ -129,6 +128,6 @@ def main() -> int: sys.exit(main()) except SystemExit: raise - except Exception as exc: # 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 print(f"[codeguard] 内部错误已忽略(fail-open): {exc!r}", file=sys.stderr) sys.exit(0) diff --git a/hooks/gate_lib.py b/hooks/gate_lib.py index ee74a7d..60d64dc 100644 --- a/hooks/gate_lib.py +++ b/hooks/gate_lib.py @@ -17,11 +17,13 @@ PLUGIN_ROOT = Path(__file__).resolve().parents[1] # hooks/ 的上级 = 插件根 sys.path.insert(0, str(PLUGIN_ROOT / "scripts")) -from detect_lang import ( # noqa: E402 - LANG_COMMANDS, detect_languages, load_user_config, project_uses_linter, probe_toolchain, +from detect_lang import ( + LANG_COMMANDS, + detect_languages, + probe_toolchain, + project_uses_linter, ) - # === git 提交内容安全检查:绝不该进版本库的文件 === # 目录(路径任一段落匹配即违规):依赖/虚拟环境/构建产物/IDE/缓存 GUARD_EXCLUDE_DIRS = { @@ -43,7 +45,7 @@ def _git(project_root: Path, *args: str) -> str | None: """跑只读 git 命令;失败(非 git 仓/无 upstream 等)返回 None,由调用方降级""" try: proc = subprocess.run( - ["git", *args], cwd=project_root, capture_output=True, text=True, timeout=15 + ["git", *args], cwd=project_root, capture_output=True, check=False, text=True, timeout=15 ) except (subprocess.TimeoutExpired, OSError, FileNotFoundError): return None @@ -101,7 +103,8 @@ def format_safety_report(violations: list[tuple[str, str, str]]) -> str: def _gate_cache_path(project_root: Path) -> Path: - import hashlib, tempfile + import hashlib + import tempfile key = hashlib.sha1(str(project_root.resolve()).encode()).hexdigest()[:12] return Path(tempfile.gettempdir()) / f"codeguard-gate-{os.getuid()}-{key}.json" @@ -122,8 +125,11 @@ def _gate_cache_key(project_root: Path, languages: list) -> str | None: GATE_CACHE_TTL = 60 # 秒:UPS 软门禁与紧随的 PreToolUse 硬门禁之间复用 -def run_gate(project_root: Path, cfg: dict) -> tuple[list, list]: - """运行全量 linter 门禁(跨进程结果缓存 + 并行执行)。 +def run_gate(project_root: Path, cfg: dict, languages: list | None = None) -> tuple[list, list]: + """运行 linter 门禁(跨进程结果缓存 + 并行执行)。 + + `languages` 可选:调用方(UserPromptSubmit)可传入用户消息里提到的 + 语言子集,门禁只跑该子集;不传则按 `detect_languages()` 全量探测。 软门禁(UserPromptSubmit)与硬门禁(PreToolUse)在正常提交路径上 会对同一状态连跑两次全量 lint——缓存键含 HEAD 与暂存区指纹, @@ -134,7 +140,8 @@ def run_gate(project_root: Path, cfg: dict) -> tuple[list, list]: - skipped: [str] 无法验证的说明(工具未装/超时),不阻塞 """ import time as _time - languages = detect_languages(project_root) + if languages is None: + languages = detect_languages(project_root) if not languages: return [], [] enabled = cfg.get("enabled_languages", []) @@ -187,7 +194,7 @@ def check(lang: str): return (lang, None, f"{lang} 项目未接入(缺 linter 配置文件),本次未验证") try: proc = subprocess.run( - gate_cmd, cwd=project_root, capture_output=True, text=True, timeout=timeout + gate_cmd, cwd=project_root, capture_output=True, check=False, text=True, timeout=timeout ) except subprocess.TimeoutExpired: return (lang, None, f"{lang} 检查超时(>{timeout}s),本次未验证") @@ -213,7 +220,7 @@ def check(lang: str): for r in results: # pool.map 保序,输出与语言表顺序一致 if r is None: continue - lang, failure, skip = r + _, failure, skip = r if failure: failures.append(failure) if skip: diff --git a/hooks/post_tool_lint.py b/hooks/post_tool_lint.py old mode 100755 new mode 100644 index 8700a0f..4eee8a6 --- a/hooks/post_tool_lint.py +++ b/hooks/post_tool_lint.py @@ -133,7 +133,7 @@ def is_ejs_template(file_path: str) -> bool: def run(cmd: list[str], cwd: Path, timeout: int = 300) -> tuple[int, str, str]: try: - proc = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout) + proc = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout, check=False) return proc.returncode, proc.stdout, proc.stderr except subprocess.TimeoutExpired: return 124, "", f"timeout after {timeout}s" @@ -293,6 +293,6 @@ def materialize(cmd: list[str]) -> list[str]: sys.exit(main()) except SystemExit: raise - except Exception as exc: # 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open:traceback 绝不进 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 old mode 100644 new mode 100755 index 3615aa9..dc7dec8 --- a/hooks/pre_tool_git_guard.py +++ b/hooks/pre_tool_git_guard.py @@ -11,8 +11,8 @@ from __future__ import annotations import json -import subprocess import os +import subprocess import sys from pathlib import Path @@ -20,8 +20,8 @@ sys.path.insert(0, str(PLUGIN_ROOT / "scripts")) sys.path.insert(0, str(PLUGIN_ROOT / "hooks")) -from detect_lang import ensure_user_path, find_project_root, load_user_config # noqa: E402 -from gate_lib import ( # noqa: E402 +from detect_lang import ensure_user_path, load_user_config +from gate_lib import ( check_commit_safety, format_safety_report, gate_directive, @@ -51,7 +51,7 @@ def _is_git_repo(p: Path) -> bool: try: proc = subprocess.run( ["git", "-C", str(p), "rev-parse", "--git-dir"], - capture_output=True, text=True, timeout=10, + capture_output=True, check=False, text=True, timeout=10, ) except (subprocess.TimeoutExpired, OSError): return False @@ -113,7 +113,7 @@ def skip_gate_via_git_config(project_root: Path) -> bool: try: proc = subprocess.run( ["git", "config", "--get", "codeguard.skipGate"], - cwd=project_root, capture_output=True, text=True, timeout=10, + cwd=project_root, capture_output=True, check=False, text=True, timeout=10, ) except (subprocess.TimeoutExpired, OSError): return False @@ -165,6 +165,6 @@ def main() -> int: sys.exit(main()) except SystemExit: raise - except Exception as exc: # 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 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 6b83c9f..b7fe0c3 100755 --- a/hooks/stop_summary.py +++ b/hooks/stop_summary.py @@ -57,6 +57,6 @@ def main() -> int: sys.exit(main()) except SystemExit: raise - except Exception as exc: # 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 print(f"[codeguard] 内部错误已忽略(fail-open): {exc!r}", file=sys.stderr) sys.exit(0) diff --git a/hooks/user_prompt_validator.py b/hooks/user_prompt_validator.py index 4107b7f..afdd064 100755 --- a/hooks/user_prompt_validator.py +++ b/hooks/user_prompt_validator.py @@ -19,8 +19,19 @@ sys.path.insert(0, str(PLUGIN_ROOT / "scripts")) sys.path.insert(0, str(PLUGIN_ROOT / "hooks")) -from detect_lang import detect_languages, ensure_user_path, find_project_root, load_user_config # noqa: E402 -from gate_lib import ( # noqa: E402 +# 触发此钩子的关键词(中英) +# 英文按词边界匹配(防止 pushed/deployment 误命中),中文子串即可。 +# 2026-09-22 实测回归:`我刚才 pushed 了`、`关于 deployment 策略的讨论` +# 必须静默(子串匹配曾把它们当提交意图跑全仓 lint)。 +import re as _re + +from detect_lang import ( + detect_languages, + ensure_user_path, + find_project_root, + load_user_config, +) +from gate_lib import ( check_commit_safety, format_safety_report, gate_directive, @@ -28,10 +39,19 @@ summarize_failures, ) -# 触发此钩子的关键词(中英) 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, +) def notify(title: str, message: str) -> None: @@ -57,12 +77,41 @@ def notify(title: str, message: str) -> None: def is_trigger(user_text: str) -> bool: + """触发条件:(1) 命中触发词(英文按词边界)(2) 非疑问句 (3) 行动意图或句首祈使。 + + (3) 缺一不可:`我刚才 pushed 了` 有触发词但无行动意图 → 静默; + `commit this now` 句首祈使 → 触发;`请帮我 commit` 行动短语 → 触发。 + """ if not user_text: return False - text = user_text.lower() - if not any(p in text for p in TRIGGER_PATTERNS): + if any(q in user_text for q in QUESTION_MARKERS): return False - return not any(q in user_text for q in QUESTION_MARKERS) + m = _TRIGGER_RE.search(user_text) + if not m: + return False + # 句首祈使:触发词本身(或前面只有空白/常见介词)位于句首。 + stripped = user_text.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_user_text() -> str: @@ -86,7 +135,10 @@ def main() -> int: cfg = load_user_config() project_root = find_project_root(os.getcwd()) or Path(os.getcwd()) - failures, skipped = run_gate(project_root, cfg) + # 2.2: 消息里提到了具体语言时只跑子集;没提到则回退全量探测(2.3)。 + detected = detect_languages(project_root) + subset = _detect_languages_in_text(user_text, detected) + failures, skipped = run_gate(project_root, cfg, languages=(subset or None)) # 提交内容安全检查:.venv/node_modules/.env/密钥等不应入库 violations = check_commit_safety(project_root, "commit") @@ -131,6 +183,6 @@ def main() -> int: sys.exit(main()) except SystemExit: raise - except Exception as exc: # 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 + except Exception as exc: # noqa: BLE001 — 内部错误 fail-open:traceback 绝不进 AI 上下文/阻断工作流 print(f"[codeguard] 内部错误已忽略(fail-open): {exc!r}", file=sys.stderr) sys.exit(0) diff --git a/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/.openspec.yaml b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/.openspec.yaml new file mode 100644 index 0000000..0e295d8 --- /dev/null +++ b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/.openspec.yaml @@ -0,0 +1,3 @@ +schema: spec-driven +created: 2026-09-22 +goal: 修复提交门禁触发启发式的过度敏感,让 MCP server 真正可用,并改进 linter 失败信息的可读性。 diff --git a/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/README.md b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/README.md new file mode 100644 index 0000000..2fd1547 --- /dev/null +++ b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/README.md @@ -0,0 +1,3 @@ +# fix-gate-trigger-and-mcp + +修复 codeguard-plugin UserPromptSubmit 钩子的触发启发式过于敏感、让 `--mcp` 入口真正可用、并改进 linter 失败信息的可读性。 diff --git a/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/design.md b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/design.md new file mode 100644 index 0000000..9863daf --- /dev/null +++ b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/design.md @@ -0,0 +1,51 @@ +## Context + +本变更的判定来自 2026-09-22 codeguard-plugin 实测排错:同一 session 里 codeguard-plugin 0.6.5 已发布,被消费的 plugin(stitch-design-plugin、codeguard-skills 等)作为基础设施出错时的第一道防线。 + +- **触发启发式现状**(2026-09-22 实测复核):`is_trigger()` 用**子串**匹配触发词;`QUESTION_MARKERS` 已同时覆盖 `?` 与 `?`。因此纯英文问句 `Should I commit this?` 本来就被排除。**真实漏洞是子串匹配**:`pushed` 命中 `push`、`deployment` 命中 `deploy`——不含行动意图的动词变体照样跑全仓 lint。修复方向是词边界 + 行动意图组合,而非再扩问号表。 +- **MCP 现状**:`scripts/run_check.py:115-123` 写 "MCP server mode not yet wired with official SDK."。`commands/check.json` 和 `commands/fix.json` 提示词里只指引 `python3 scripts/run_check.py`,没有提 MCP。意味着 MCP 入口是"摆设",agent 唯一可用路径仍是 shell 子进程。 +- **失败输出现状**:`scripts/run_check.py:96` 取 stderr 最后 3 行的第 1 行再截 120 字符。实测 ruff 对单条违规的描述就超过 120 字符(路径 + 行号 + 错误码 + 描述),截断后用户得重跑或加 `--output-format=concise` 才能定位。 + +## Goals / Non-Goals + +**Goals:** + +- 让"使用过程中的情况"这种含"发布"二字但意图为询问的句子不再误触发全仓 lint。 +- 让英文问句(带 `?`)同样被识别为非提交意图。 +- 触发后只跑用户消息中实际提到的语言(提速 + 减少噪声)。 +- `--mcp` 走官方 `mcp` Python SDK,暴露三个工具;agent 不再被强制 spawn 子进程。 +- 失败时 stderr 完整写入磁盘,终端只给摘要。 + +**Non-Goals:** + +- 不修改 `commands/check.json` / `commands/fix.json` 的提示词结构(已存在的命令可继续工作;仅在 README 里增加 MCP 用法说明)。 +- 不修改 vendored 技能(codeguard-* 走 vendor 流程;本变更涉及的语言与脚本都是仓内自有代码)。 +- 不重写 `commands/init.json` / `assets/templates/pre-commit-config.yaml`(与本变更无直接关系,且已在前置审计中标记为独立改进项)。 +- 不引入自动 commit / 自动 push——`/check` 与 `/fix` 仍只读 + 只修,提交门禁仍由 PreToolUse 钩子把守。 + +## Decisions + +1. **触发判定改为词边界 + 行动意图双条件**:先 `re.search(r"\b(commit|push|deploy)\b|提交|发布|部署", text, re.I)`(中文词无边界问题,按原样保留子串;英文词加 `\b`)。再要求同时命中行动意图:`re.search(r"(请帮我|帮我|请|马上|立刻|现在|下一步|继续|please|now|next|proceed|go ahead|commit this|push this|deploy this)", text, re.I)` **或** 触发词位于句首(祈使语气,如 `commit this now`)。两者其一即进入 lint 路径;否则静默退出 0。`QUESTION_MARKERS` 仅加回归用例,不改内容。 +2. **英文 `?` 加入 `QUESTION_MARKERS`**:把 `?` 加入列表,并把"问句句尾 `?` 紧跟在触发词后"做特例豁免(提问「要不要 commit 这个?」即使包含 commit 也不进入门禁)。 +3. **trigger 后只跑检测到的语言**:从用户消息中正则提取候选语言 token(与 `languages.json` 的 `id` 字段做大小写无关匹配),命中的子集运行;没有命中则 fallback 到 `detect_languages()` 全量(保持原有探测能力)。 +4. **`--mcp` 走官方 SDK**:使用 `mcp.server.Server` + `stdio_server()` 注册三个工具: + - `check_code_style(path, languages?)`:跑 lint,返回 `{language, passed, exit_code, stderr_path}` 结构。 + - `auto_fix(path, languages?)`:调 formatter,再跑 lint 验证。 + - `list_languages()`:返回 `languages.json` 的 `id` 与 `name`,便于 agent 端校验输入。 + 删掉占位 `mcp_main` 分支;新增 `requirements-test.txt` 引入 `mcp>=1.0`。 +5. **失败信息落盘**:`scripts/run_check.py` 增加一个可选 `--log-dir` 参数(默认 `out/`),将完整 stderr 写到 `out/.codeguard-last.log`;终端输出改为"完整日志:"。保留 `--quiet` 反向开关用于 CI。 + +## Risks / Trade-offs + +- **触发判定变严后真触发是否会漏判?**:在排错里手工构造 12 条真实用户消息("请帮我 commit"、"commit this" 等)逐条验证。漏判成本 = 用户主动提交时不被门禁拦下,违反 #1 优先级(保安全)。需要红蓝两套回归。 +- **MCP 引入新依赖**:`mcp` SDK 是 1.0+ 包,依赖较重。在 vendored 技能外,仓内 `requirements-test.txt` 增加一行 `mcp>=1.0`;CI 安装会自动取到。 +- **log 文件泄漏**:写到 `out/.codeguard-last.log` 默认随仓库 `.gitignore`(`out/` 已忽略),不污染版本库;若仓库没忽略 `out/`,本变更需在 `design.md` Migration Plan 加一行 `.gitignore` 检查。 + +## Migration Plan + +1. 修改 `hooks/user_prompt_validator.py`:扩展 `TRIGGER_PATTERNS` / `QUESTION_MARKERS`,新增 `_detect_languages_in_text` 辅助函数,在 `main()` 中按子集运行门禁。 +2. 重写 `scripts/run_check.py` 的失败输出逻辑与 `mcp_main()` 实现。 +3. 在 `requirements-test.txt` 增加 `mcp>=1.0`。 +4. 在 README.md 与 README.zh-CN.md 补 MCP 用法一段(不修改 commands/check.json 本身)。 +5. 跑 `python3 tests/run_all.py`(141 项)+ `python3 -m unittest discover -s tests -p 'test_*.py'` 全绿。 +6. `openspec validate --strict` 通过后按 `AGENTS.md` 强制发版规则 bump minor 版本(MCP 服务器属新功能:0.6.6 → 0.7.0),归档本变更。 diff --git a/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/proposal.md b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/proposal.md new file mode 100644 index 0000000..6d4c12e --- /dev/null +++ b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/proposal.md @@ -0,0 +1,31 @@ +## Why + +本变更解决三个在真实使用中持续踩中的痛点: + +1. **提交门禁触发过于敏感(子串匹配)**:`hooks/user_prompt_validator.py` 的 `is_trigger()` 对 `TRIGGER_PATTERNS = ["commit", "push", "deploy", "提交", "发布", "部署"]` 做**子串**匹配。实测(2026-09-22)真实误触发样例:`我刚才 pushed 了`(命中 `push`)、`关于 deployment 策略的讨论`(命中 `deploy`)——历史行为记录在该文件 `QUESTION_MARKERS` 的注释里(`「git 提交和推送,是不能把 .venv 排除掉么」被当成提交意图`)。纯词边界匹配可消除这类误触发;同时 `?` 已在 `QUESTION_MARKERS` 中,但**触发词与行动意图的组合判定**缺失,导致"非提交语境里的动词"照样进全仓门禁。 +2. **`--mcp` 是占位**:`scripts/run_check.py:115-123` 的 `mcp_main()` 明文写"MCP server mode not yet wired with official SDK"。文档承诺的 MCP server 入口实际不可用,AI agent 想在不改 shell spawn 的前提下复用 codeguard 失败,只能重新发明。 +3. **失败输出截断**:`scripts/run_check.py:96` 失败时仅 `stderr.splitlines()[-3:][0][:120]`,路径 + 错误码 + 长描述常超过 120 字符,截断后用户得自己翻完整日志。 + +## What Changes + +- 触发启发式改为词边界匹配 + 行动意图组合判定(触发词必须出现在祈使/请求语境,而非任意子串),trigger 后只跑"当前消息提到的语言"而非 `detect_languages()` 的全量。`QUESTION_MARKERS` 保留现有中英双标点并加回归用例锁定。 +- `--mcp` 走官方 `mcp` Python SDK:注册 `check_code_style` / `auto_fix` / `list_languages` 三个 stdio JSON-RPC 工具,删除占位分支。 +- linter 失败:完整 stderr 写入 `out/.codeguard-last.log`,终端只显示问题摘要与日志路径。 + +## Capabilities + +### New Capabilities + +- `gate-trigger-policy`: Defines when the UserPromptSubmit hook runs the lint gate and which languages it checks; explicitly documents when it stays silent. +- `mcp-tool-server`: Defines the stdio JSON-RPC surface exposed by `scripts/run_check.py --mcp`, the three tools, their inputs and outputs, and the failure envelope. + +### Modified Capabilities + +None. + +## Impact + +- `hooks/user_prompt_validator.py`:触发判定与语言选择逻辑改动;外部行为变化仅在误触发场景(更少误触发为正向变化)。 +- `scripts/run_check.py`:新增官方 SDK 依赖(`mcp` 包需写入 `requirements.txt` 或类似清单),新增 `out/.codeguard-last.log` 落盘逻辑,删除占位 `mcp_main` 分支。 +- 新建根 `requirements.txt` 声明 `mcp>=1.0`(本仓当前没有任何 requirements 文件;`mcp` 1.26.0 已在开发环境安装,CI 需显式安装)。 +- 不修改 vendored 技能(codeguard-* 走 vendor 流程);不修改 marketplace 清单。 diff --git a/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/specs/gate-trigger-policy/spec.md b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/specs/gate-trigger-policy/spec.md new file mode 100644 index 0000000..2e88de6 --- /dev/null +++ b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/specs/gate-trigger-policy/spec.md @@ -0,0 +1,40 @@ +## ADDED Requirements + +### Requirement: The UserPromptSubmit hook SHALL run the lint gate only when intent is commit-like +The `UserPromptSubmit` hook MUST evaluate the user's prompt for commit intent using word-boundary matching of trigger words (`commit|push|deploy|发布|部署`) and a separate action-verb check. The hook SHALL exit silently when either (a) no trigger word is present, (b) the only trigger occurrence is followed by `?`, or (c) a trigger word appears in an interrogative sentence identified by `QUESTION_MARKERS` (`?`, `?`, `么`, `吗`, `如何`, `怎么`, `有没有`, `是不是`, `什么是`, `哪些`). + +#### Scenario: User asks a question containing a trigger word + +- **WHEN** the user prompt is `使用过程中的情况` or `Should I commit this?` or `先读项目结构` +- **THEN** the hook exits 0 with empty stdout; no linter runs + +#### Scenario: User issues a commit-like imperative + +- **WHEN** the user prompt is `请帮我 commit` or `commit this now` or `请帮我 push` +- **THEN** the hook enters the gate path and invokes `run_gate()` + +#### Scenario: Trigger word followed by question mark is exempt + +- **WHEN** the user prompt is `commit?` or `要不要 push?` or `should I deploy?` +- **THEN** the hook exits silently regardless of other QUESTION_MARKERS content + +### Requirement: The gate SHALL run only languages named in the user message when they can be identified +When the trigger fires, the hook MUST extract candidate language ids from the user prompt using case-insensitive token matching against `scripts/languages.json` ids. When the intersection is non-empty, the gate SHALL run only that subset. When the intersection is empty, the gate SHALL fall back to the existing `detect_languages()` behavior. + +#### Scenario: User mentions one language explicitly + +- **WHEN** the user prompt is `请帮我 commit 这次 ruff 改动` +- **THEN** the gate runs only Python; other detected languages are skipped for this invocation + +#### Scenario: User mentions no language + +- **WHEN** the user prompt is `请帮我 commit` with no language token +- **THEN** the gate runs every language detected by `detect_languages()` (current behavior preserved) + +### Requirement: Question markers SHALL cover both Chinese and English +`QUESTION_MARKERS` MUST include the ASCII `?` in addition to the existing Chinese markers, so that an English prompt ending with `?` is recognized as a question even without other markers. + +#### Scenario: ASCII question mark without any other marker + +- **WHEN** the user prompt is `Should I push this?` +- **THEN** the hook exits silently because `?` is a recognized question marker diff --git a/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/specs/mcp-tool-server/spec.md b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/specs/mcp-tool-server/spec.md new file mode 100644 index 0000000..530aaf6 --- /dev/null +++ b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/specs/mcp-tool-server/spec.md @@ -0,0 +1,43 @@ +## ADDED Requirements + +### Requirement: The MCP server SHALL expose three tools over stdio JSON-RPC +`scripts/run_check.py --mcp` MUST start an stdio JSON-RPC server using the official `mcp` Python SDK and register exactly three tools: `check_code_style`, `auto_fix`, and `list_languages`. The server MUST NOT take any other CLI flag in MCP mode. + +#### Scenario: Listing tools at startup + +- **WHEN** a client sends `tools/list` +- **THEN** the response enumerates `check_code_style`, `auto_fix`, `list_languages` with their input schemas + +### Requirement: `check_code_style` SHALL return a structured envelope with full stderr on disk +`check_code_style(path, languages?)` MUST run the configured linters, write the full stderr of any failing invocation to `/.codeguard-last.log` (default `/out/.codeguard-last.log`), and return a JSON envelope with one entry per language: `{language, passed: bool, exit_code: int, stderr_path: str, log_path: str}`. + +#### Scenario: One language fails + +- **WHEN** the path contains a single Python project whose ruff lint fails +- **THEN** the envelope contains one entry with `passed=false`, `exit_code` from ruff, `stderr_path` pointing at the on-disk log, and `log_path` echoing the same path + +#### Scenario: All languages pass + +- **WHEN** every linter exits 0 +- **THEN** every entry has `passed=true`, `exit_code=0`, and the log file is either empty or absent (no stderr written) + +### Requirement: `auto_fix` SHALL run formatters and re-verify +`auto_fix(path, languages?)` MUST invoke the per-language formatter chain (`format` from `scripts/languages.json`) and then re-run `check_code_style` on the same scope before returning. The response envelope MUST contain both `fixed: bool` (whether any file was modified) and the `check_code_style` envelope for the post-fix re-check. + +#### Scenario: Formatter succeeds and lint passes + +- **WHEN** ruff `--fix` removes unused imports and the re-check passes +- **THEN** `fixed=true` and the embedded `check_code_style` envelope is all-pass + +#### Scenario: Formatter succeeds but lint still fails + +- **WHEN** ruff `--fix` ran but a remaining error is non-fixable +- **THEN** `fixed=true` and the embedded `check_code_style` envelope contains at least one failing entry + +### Requirement: `list_languages` SHALL return id + name only +`list_languages()` MUST return a JSON array of `{id, name}` objects from `scripts/languages.json`, omitting the rest of each language record (lint command, format command, install hint, etc.) so internal commands are not exposed as part of the MCP surface. + +#### Scenario: Listing all supported languages + +- **WHEN** the client calls `list_languages` +- **THEN** the response enumerates every entry in `scripts/languages.json` with exactly two fields (`id`, `name`) per record diff --git a/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/tasks.md b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/tasks.md new file mode 100644 index 0000000..8516706 --- /dev/null +++ b/openspec/changes/2026-09-22-fix-gate-trigger-and-mcp/tasks.md @@ -0,0 +1,48 @@ +## 1. Trigger heuristics + +- [x] 1.1 Replace substring `in` match with word-boundary match for English trigger words (`\b(commit|push|deploy)\b`) in `hooks/user_prompt_validator.py`; keep Chinese trigger words (`提交|发布|部署`) as substring (no boundary semantics needed). +- [x] 1.2 Add action-intent detection: a message passes only when the trigger word is paired with an action-intent phrase (`请帮我|帮我|请|马上|立刻|现在|下一步|继续|please|now|next|proceed|go ahead|commit this|push this|deploy this`) **or** the trigger word stands at the start of the message (bare imperative). +- [x] 1.3 Add regression tests locking the existing `QUESTION_MARKERS` content (both `?` and `?` already present) — content itself is unchanged. +- [x] 1.4 Add regression tests for sentence-initial bare imperatives (`commit this now`, `提交代码`) that must still trigger. + +## 2. Targeted language detection + +- [x] 2.1 Add `_detect_languages_in_text(text, language_ids)` helper that returns the intersection of `re.findall(r"\w+", text.lower())` with `languages.json` ids. +- [x] 2.2 When the helper returns a non-empty set, restrict `run_gate()` to that subset. +- [x] 2.3 When the helper returns empty, fall back to existing `detect_languages()` semantics. + +## 3. MCP server + +- [x] 3.1 Add root `requirements.txt` with `mcp>=1.0,<2` (mcp 2.x changed the decorator API; pin to the tested major) and install it in skills-check.yml before unittest. +- [x] 3.2 Replace `scripts/run_check.py:mcp_main()` with a real `mcp.server.Server` that registers `check_code_style`, `auto_fix`, `list_languages` tools over stdio JSON-RPC. +- [x] 3.3 Ensure `check_code_style` returns a structured `{language, passed, exit_code, stderr_path}` payload and writes full stderr to `out/.codeguard-last.log`. +- [x] 3.4 Ensure `auto_fix` invokes the formatter chain and re-runs `check_code_style` before returning. +- [x] 3.5 Ensure `list_languages` returns `languages.json` ids + names without exposing internal schema. + +## 4. Failure log + +- [x] 4.1 Add `--log-dir` flag (default `out/`) to `scripts/run_check.py`. +- [x] 4.2 When a language fails, append full stderr to `/.codeguard-last.log` and emit only a summary line plus the absolute log path to stdout. +- [x] 4.3 Honor `--quiet` to suppress the log write and summary line for CI runs. + +## 5. Documentation + +- [x] 5.1 Append a "MCP server" subsection to `README.md` with one CLI invocation example. +- [x] 5.2 Mirror it in `README.zh-CN.md`. +- [x] 5.3 Confirm `.gitignore` covers `out/` so the on-disk log never enters the index. + +## 6. Verification + +- [x] 6.1 Add unit tests covering: trigger does NOT fire on `我刚才 pushed 了`, `关于 deployment 策略的讨论`, `Should I commit this?`; trigger fires on `请帮我 commit`, `commit this now`, `请帮我 push`, `提交代码`. +- [x] 6.2 Add a unit test that `_detect_languages_in_text` narrows the gate to a subset (e.g. prompt says `commit this rust change` → only `rust` runs). +- [x] 6.3 Add a unit test that `--mcp` boots, lists tools, and returns a tool result. +- [x] 6.4 Run `python3 -m unittest discover -s tests -p 'test_*.py'` — all green. +- [x] 6.5 Run `python3 scripts/validate_skills.py skills` — unchanged. +- [x] 6.6 Run `openspec validate --all --strict`. + +## 7. Release + +- [ ] 7.1 `node scripts/bump-plugin.mjs codeguard minor` (0.6.6 → 0.7.0; MCP server is a new feature). +- [ ] 7.2 Sync marketplace manifest + hub catalog. +- [ ] 7.3 Push, confirm CI green, tag v0.7.0, GitHub Release. +- [ ] 7.4 `openspec archive 2026-09-22-fix-gate-trigger-and-mcp`. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1665c74 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +mcp>=1.0,<2 diff --git a/scripts/cve_check.py b/scripts/cve_check.py index 95f9163..d34ef6c 100644 --- a/scripts/cve_check.py +++ b/scripts/cve_check.py @@ -49,7 +49,7 @@ def severities_at_and_above(threshold: str) -> list[str]: def run(cmd: list[str], cwd: Path, timeout: int = 600) -> tuple[int, str, str]: try: - proc = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout) + proc = subprocess.run(cmd, cwd=cwd, capture_output=True, check=False, text=True, timeout=timeout) return proc.returncode, proc.stdout, proc.stderr except subprocess.TimeoutExpired: return 124, "", f"timeout after {timeout}s" diff --git a/scripts/detect_lang.py b/scripts/detect_lang.py old mode 100755 new mode 100644 index 39883d9..0aa03d3 --- a/scripts/detect_lang.py +++ b/scripts/detect_lang.py @@ -186,7 +186,7 @@ def _cached(key: tuple, fn): def _run_probe_cmd(probe: list, timeout: int) -> tuple[bool, str]: try: # stdin=DEVNULL:探活命令绝不消费宿主 stdin;--format 类探活靠 EOF 立即返回 - proc = subprocess.run(probe, capture_output=True, text=True, timeout=timeout, + proc = subprocess.run(probe, capture_output=True, check=False, text=True, timeout=timeout, stdin=subprocess.DEVNULL) except subprocess.TimeoutExpired: return False, f"探活超时: {' '.join(probe)}" @@ -203,7 +203,7 @@ def _probe_binary(b: str, timeout: int) -> tuple[bool, str]: if shutil.which(b) is None: return False, f"{b} 不在 PATH" try: - proc = subprocess.run([b, "--version"], capture_output=True, text=True, timeout=timeout) + proc = subprocess.run([b, "--version"], capture_output=True, check=False, text=True, timeout=timeout) except subprocess.TimeoutExpired: return False, f"{b} --version 超时" except OSError as exc: diff --git a/scripts/dockerfile_security.py b/scripts/dockerfile_security.py index f8aaeec..9c195f7 100644 --- a/scripts/dockerfile_security.py +++ b/scripts/dockerfile_security.py @@ -27,7 +27,7 @@ def run(cmd: list[str], cwd: Path, timeout: int = 300) -> tuple[int, str, str]: try: - proc = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout) + proc = subprocess.run(cmd, cwd=cwd, capture_output=True, check=False, text=True, timeout=timeout) return proc.returncode, proc.stdout, proc.stderr except subprocess.TimeoutExpired: return 124, "", f"timeout after {timeout}s" @@ -56,16 +56,15 @@ def hadolint_scan(files: list[Path], root: Path) -> dict: tool_ok = True scanned = 0 for f in files: - rc, out, err = run(["hadolint", str(f)], cwd=root) + rc, out, _err = run(["hadolint", str(f)], cwd=root) if rc == 127: return {"tool": "hadolint", "exit": 127, "findings": [], "summary_tail": "hadolint not installed (brew install hadolint)"} if rc not in (0, 1): # hadolint: 0=clean, 1=has findings tool_ok = False scanned += 1 - for line in out.splitlines(): - # 格式: file:line DLxxxx severity message - findings.append(line) + # 格式: file:line DLxxxx severity message + findings.extend(out.splitlines()) return {"tool": "hadolint", "exit": 0 if not findings else 2, "scanned": scanned, "findings": findings[-40:], "tool_ok": tool_ok} @@ -74,7 +73,7 @@ def trivy_scan(files: list[Path], root: Path) -> dict: """trivy config:misconfig 规则(root 运行、digest 未固定、secrets 等)""" results = [] for f in files: - rc, out, err = run(["trivy", "config", "--format", "json", "--quiet", str(f)], + rc, out, _err = run(["trivy", "config", "--format", "json", "--quiet", str(f)], cwd=root, timeout=600) if rc == 127: return {"tool": "trivy config", "exit": 127, "findings": [], diff --git a/scripts/fix.py b/scripts/fix.py old mode 100755 new mode 100644 diff --git a/scripts/gen_language_docs.py b/scripts/gen_language_docs.py old mode 100755 new mode 100644 diff --git a/scripts/paths.py b/scripts/paths.py index 20702a3..d4b4509 100644 --- a/scripts/paths.py +++ b/scripts/paths.py @@ -70,7 +70,7 @@ def _node_available() -> bool: shell = os.environ.get("SHELL") or "/bin/zsh" proc = subprocess.run( [shell, "-lc", "printf '%s' \"$PATH\""], - capture_output=True, text=True, timeout=5, + capture_output=True, check=False, text=True, timeout=5, ) if proc.returncode == 0: inherited = proc.stdout.strip().splitlines() diff --git a/scripts/run_check.py b/scripts/run_check.py old mode 100755 new mode 100644 index 281a80c..50b9c93 --- a/scripts/run_check.py +++ b/scripts/run_check.py @@ -4,25 +4,45 @@ python3 run_check.py # 检测当前项目语言并全跑 python3 run_check.py --lang java # 只跑 java python3 run_check.py --fix # 失败时自动修复 - python3 run_check.py --mcp # MCP server 模式 - -按语言执行细节(subprocess 调用、退出码归一化、fix 复检)全部委托给 -`scripts/run_per_language.py`;本文件仅负责 argparse + 报告输出。 + python3 run_check.py --log-dir out # 失败时完整输出落盘(默认 /out) + python3 run_check.py --quiet # 不写日志、不打摘要行(CI 用) + python3 run_check.py --mcp # MCP server 模式(官方 SDK stdio) + +MCP 模式(OpenSpec 2026-09-22-fix-gate-trigger-and-mcp): + 暴露 check_code_style / auto_fix / list_languages 三个工具; + 依赖 `mcp>=1.0`(见根 requirements.txt)。按语言执行细节全部委托给 + `scripts/run_per_language.py`;本文件负责 argparse、报告输出与 MCP 接线。 """ from __future__ import annotations import argparse +import asyncio +import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import run_per_language from detect_lang import ( + REGISTRY_PATH, detect_languages, find_project_root, load_user_config, ) +LOG_NAME = ".codeguard-last.log" + + +def _filter_languages(project_root: Path, lang_arg: str | None) -> list[str]: + cfg = load_user_config() + enabled = cfg.get("enabled_languages", []) + languages = detect_languages(project_root) + if enabled and enabled != ["auto"]: + languages = [lang for lang in languages if lang in enabled] + if lang_arg: + languages = [lang for lang in languages if lang in lang_arg.split(",")] + return languages + def cli_main(): parser = argparse.ArgumentParser() @@ -30,6 +50,10 @@ def cli_main(): parser.add_argument("--fix", action="store_true", help="失败时尝试自动修复") parser.add_argument("--mcp", action="store_true", help="启动 MCP server(stdio)") parser.add_argument("--timeout", type=int, default=120) + parser.add_argument("--log-dir", default=None, + help="失败时完整输出落盘目录(默认 /out;--quiet 时不写)") + parser.add_argument("--quiet", action="store_true", + help="不写日志、不打印完整日志摘要行(CI 用)") parser.add_argument("path", nargs="?", default=".", help="项目根") args = parser.parse_args() @@ -38,14 +62,13 @@ def cli_main(): if args.mcp: return mcp_main(project_root) - cfg = load_user_config() - enabled = cfg.get("enabled_languages", []) - languages = detect_languages(project_root) - if enabled and enabled != ["auto"]: - languages = [lang for lang in languages if lang in enabled] - if args.lang: - languages = [lang for lang in languages if lang in args.lang.split(",")] + log_dir: Path | None + if args.quiet: + log_dir = None + else: + log_dir = Path(args.log_dir) if args.log_dir else project_root / "out" + languages = _filter_languages(project_root, args.lang) if not languages: print("[codeguard] 未识别到语言", file=sys.stderr) return 1 @@ -56,36 +79,130 @@ def cli_main(): results = run_per_language.run_check( languages, project_root, - timeout=args.timeout, fix=args.fix, + timeout=args.timeout, fix=args.fix, log_dir=log_dir, ) for r, lang in zip(results, languages): if r.get("dry_run"): continue if not r["passed"] and args.fix: - print(f"[codeguard] \u26a0\ufe0f {lang} lint failed, attempting auto-fix...") - status = "\u2705 passed" if r["passed"] else f"\u274c FAILED (exit={r.get('exit_code')})" + print(f"[codeguard] ⚠️ {lang} lint failed, attempting auto-fix...") + status = "✅ passed" if r["passed"] else f"❌ FAILED (exit={r.get('exit_code')})" print(f" {lang:12s} {status}") - if not r["passed"] and r.get("stderr_tail"): - print(f" \u2514\u2500 {r['stderr_tail'].splitlines()[-3:][0][:120]}") + if not r["passed"] and r.get("log_path") and not args.quiet: + print(f" └─ 完整日志: {r['log_path']}") print() if all(r["passed"] for r in results): - print("[codeguard] \u2705 all passed") + print("[codeguard] ✅ all passed") return 0 - print("[codeguard] \u274c some lints failed") + print("[codeguard] ❌ some lints failed") return 2 -def mcp_main(project_root: Path): - """极简 MCP server:暴露 check_code_style 工具。 +# ══════════════════════════ MCP server(官方 SDK) ══════════════════════════ - 协议参考:https://modelcontextprotocol.io - 这里用 stdio JSON-RPC 风格——真实工程里应换成官方 SDK。 +def _mcp_envelope(results: list[dict]) -> list[dict]: + """把 run_per_language 结果收成 MCP 契约信封: + {language, passed, exit_code, stderr_path, log_path}。 + stderr_path 与 log_path 指向同一完整输出文件;通过的语言为空串。 """ - # 占位实现:调用方传 {"method": "tools/call", "params": {"name": "check_code_style"}} - # 真实落地时建议用 mcp Python SDK:https://github.com/modelcontextprotocol/python-sdk - print("[codeguard] MCP server mode not yet wired with official SDK.", file=sys.stderr) - print("[codeguard] Use CLI mode for now: python3 run_check.py", file=sys.stderr) + envelope = [] + for r in results: + log_path = r.get("log_path", "") + envelope.append({ + "language": r.get("language", ""), + "passed": bool(r.get("passed")), + "exit_code": int(r.get("exit_code", 0)), + "stderr_path": log_path, + "log_path": log_path, + }) + return envelope + + +def _load_registry_entries() -> list[dict]: + data = json.loads(REGISTRY_PATH.read_text(encoding="utf-8")) + return [ + {"id": entry["id"], "name": entry.get("name", entry["id"])} + for entry in data.get("languages", []) + ] + + +def mcp_main(project_root: Path) -> int: + """官方 mcp SDK stdio server:check_code_style / auto_fix / list_languages。""" + try: + import mcp.types as mcp_types + from mcp.server import Server + from mcp.server.stdio import stdio_server + except ModuleNotFoundError: + print("[codeguard] mcp SDK 未安装:pip install -r requirements.txt", file=sys.stderr) + return 1 + + server = Server("codeguard") + input_schema = { + "type": "object", + "properties": { + "path": {"type": "string", "description": "项目根;缺省用 --mcp 启动时给的 path"}, + "languages": {"type": "array", "items": {"type": "string"}, + "description": "只跑这些语言 id;缺省按项目自动探测"}, + }, + } + + def text_block(payload) -> mcp_types.TextContent: + return mcp_types.TextContent(type="text", text=json.dumps(payload, ensure_ascii=False)) + + @server.list_tools() + async def list_tools() -> list: + return [ + mcp_types.Tool( + name="check_code_style", + description="对项目跑 lint,返回逐语言结果信封与完整输出日志路径", + inputSchema=input_schema, + ), + mcp_types.Tool( + name="auto_fix", + description="先跑 formatter 链,再复检 lint(嵌入 check 信封)", + inputSchema=input_schema, + ), + mcp_types.Tool( + name="list_languages", + description="列出支持的语言 id 与显示名(不含内部 lint/format 命令)", + inputSchema={"type": "object", "properties": {}}, + ), + ] + + @server.call_tool() + async def call_tool(name: str, arguments: dict) -> list: + if name == "list_languages": + return [text_block(_load_registry_entries())] + if name not in ("check_code_style", "auto_fix"): + return [text_block({"error": f"unknown tool: {name}"})] + args = arguments or {} + raw = args.get("path") or str(project_root) + root = find_project_root(raw) or Path(raw).resolve() + requested = args.get("languages") + langs = list(requested) if requested else _filter_languages(root, None) + if not langs: + return [text_block({"error": "未识别到语言", "path": str(root)})] + if name == "auto_fix": + fix_results = run_per_language.run_fix(langs, root) + results = run_per_language.run_check(langs, root, log_dir=root / "out") + any_fixed = any( + r.get("fixed") for r in fix_results if not r.get("dry_run") + ) + payload = { + "fixed": bool(any_fixed), + "check": _mcp_envelope(results), + } + else: + results = run_per_language.run_check(langs, root, log_dir=root / "out") + payload = _mcp_envelope(results) + return [text_block(payload)] + + async def _serve() -> None: + async with stdio_server() as (read, write): + await server.run(read, write, server.create_initialization_options()) + + asyncio.run(_serve()) return 0 diff --git a/scripts/run_per_language.py b/scripts/run_per_language.py index 4c1c472..5ce3735 100644 --- a/scripts/run_per_language.py +++ b/scripts/run_per_language.py @@ -22,7 +22,7 @@ def _run(cmd: list[str], cwd: Path, timeout: int) -> tuple[int, str, str]: """subprocess 调用统一封装:归一化超时(124) / 命令缺失(127) / 其它返回原码。""" try: - proc = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout) + proc = subprocess.run(cmd, cwd=cwd, capture_output=True, check=False, text=True, timeout=timeout) except subprocess.TimeoutExpired: return 124, "", f"timeout after {timeout}s" except FileNotFoundError as e: @@ -32,9 +32,16 @@ def _run(cmd: list[str], cwd: Path, timeout: int) -> tuple[int, str, str]: def run_check(languages: list[str], project_root: Path, *, timeout: int = 120, fix: bool = False, - dry_run: bool = False) -> list[dict]: - """对每个语言跑 lint;fix=True 时失败后自动跑 format 复检一次。""" + dry_run: bool = False, log_dir: Path | None = None) -> list[dict]: + """对每个语言跑 lint;fix=True 时失败后自动跑 format 复检一次。 + + `log_dir` 非空且有语言失败时,完整输出(stdout+stderr 合并——ruff 等 + linter 把诊断打到 stdout)落盘到 `/.codeguard-last.log` + (组头含语言与退出码),失败条目附带 `log_path`; + 全部通过则不产生日志文件。 + """ results: list[dict] = [] + log_entries: list[tuple[str, int, str]] = [] for lang in languages: cmd_def = LANG_COMMANDS.get(lang) if not cmd_def: @@ -60,6 +67,32 @@ def run_check(languages: list[str], project_root: Path, "stderr_tail": err[-2000:] if err else "", "stdout_tail": out[-1000:] if out else "", }) + if not passed and (err or out): + combined = "" + if out: + combined += out + if err: + if combined and not combined.endswith("\n"): + combined += "\n" + combined += err + log_entries.append((lang, rc, combined)) + if log_dir is not None and log_entries: + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / ".codeguard-last.log" + chunks = [] + for lang, rc, content in log_entries: + block = f"===== {lang} (exit={rc}) =====\n{content}" + if not block.endswith("\n"): + block += "\n" + chunks.append(block) + try: + log_path.write_text("\n".join(chunks), encoding="utf-8") + except OSError: + log_path = None + if log_path is not None: + for r in results: + if not r["passed"]: + r["log_path"] = str(log_path.resolve()) return results @@ -77,7 +110,7 @@ def run_fix(languages: list[str], project_root: Path, results.append({"language": lang, "fixed": True, "dry_run": True, "command": fmt or []}) continue - rc, out, err = _run(fmt, cwd=project_root, timeout=timeout) + rc, _out, err = _run(fmt, cwd=project_root, timeout=timeout) results.append({ "language": lang, "fixed": rc == 0, diff --git a/scripts/vendor/skill_vendor.py b/scripts/vendor/skill_vendor.py old mode 100644 new mode 100755 index fab55d0..b434174 --- a/scripts/vendor/skill_vendor.py +++ b/scripts/vendor/skill_vendor.py @@ -25,11 +25,9 @@ import re import shutil import subprocess -import sys import tempfile from pathlib import Path - SKILL_NAME_RE = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$") VERSION_TAG_RE = re.compile(r"^v[0-9]+\.[0-9]+\.[0-9]+$") COMMIT_SHA_RE = re.compile(r"^[0-9a-f]{40}$") @@ -66,9 +64,7 @@ def resolve_ref(repo: str, ref: str) -> str: for line in result.splitlines(): candidate, _, remote_name = line.partition("\t") short = remote_name.removeprefix("refs/tags/").removeprefix("refs/heads/") - if short == f"{ref}^{{}}": - resolved = candidate - elif short == ref and resolved is None: + if short == f"{ref}^{{}}" or short == ref and resolved is None: resolved = candidate if resolved is None: raise RuntimeError(f"{repo}: could not resolve ref '{ref}'") @@ -138,10 +134,10 @@ def validate_plugin_local_inventory(root: Path, lock: dict) -> None: if policy.get("version") != 1: raise RuntimeError("unsupported plugin-local-skills.json version") if not isinstance(policy.get("dest"), str): - raise RuntimeError("plugin-local-skills.json requires a dest string") + raise RuntimeError("plugin-local-skills.json requires a dest string") # noqa: TRY004 names = policy.get("skills") if not isinstance(names, list): - raise RuntimeError("plugin-local-skills.json requires a skills list") + raise RuntimeError("plugin-local-skills.json requires a skills list") # noqa: TRY004 if len(names) != len(set(names)): raise RuntimeError("plugin-local-skills.json contains duplicate skill names") for name in names: diff --git a/tests/run_all.py b/tests/run_all.py old mode 100644 new mode 100755 index ed39055..f95537f --- a/tests/run_all.py +++ b/tests/run_all.py @@ -50,12 +50,12 @@ def run_hook(script: str, payload: dict | None, cwd: Path, env_extra: dict | Non return subprocess.run( [sys.executable, str(HOOKS / script)], input=json.dumps(payload) if payload is not None else "", - capture_output=True, text=True, cwd=cwd, timeout=120, env=env, + capture_output=True, check=False, text=True, cwd=cwd, timeout=120, env=env, ) def git(repo: Path, *args: str): - return subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True) + return subprocess.run(["git", *args], cwd=repo, capture_output=True, check=False, text=True) def make_repo() -> Path: @@ -77,7 +77,7 @@ def make_repo() -> Path: def test_languages(): print("\n[1] 语言规则结构审计(全部 stable/beta)") - langs = json.load(open(PLUGIN / "scripts" / "languages.json"))["languages"] + langs = json.loads((PLUGIN / "scripts" / "languages.json").read_text(encoding="utf-8"))["languages"] stable = [l for l in langs if l.get("status") in ("stable", "beta")] ok(f"stable/beta 数量 = {len(stable)}(≥50)", len(stable) >= 50) @@ -421,9 +421,9 @@ def test_edges(): cli = str(PLUGIN / "bin" / "codeguard") if os.path.exists(cli): env = {**os.environ, "PYTHONIOENCODING": "utf-8"} - r = subprocess.run([cli, "detect"], capture_output=True, text=True, cwd=repo, timeout=60, env=env) + r = subprocess.run([cli, "detect"], capture_output=True, check=False, text=True, cwd=repo, timeout=60, env=env) ok("codeguard detect 退出", r.returncode in (0, 1), f"exit={r.returncode}") - r = subprocess.run([cli, "dockerfile"], capture_output=True, text=True, cwd=repo, timeout=90, env=env) + r = subprocess.run([cli, "dockerfile"], capture_output=True, check=False, text=True, cwd=repo, timeout=90, env=env) ok("codeguard dockerfile(无 Dockerfile→未验证=1 或通过=0)", r.returncode in (0, 1), f"exit={r.returncode}") else: skip("CLI 冒烟", "bin/codeguard 不存在") @@ -535,9 +535,9 @@ def counting(root, cfg, langs): return orig_uncached(root, cfg, langs) gate_lib._run_gate_uncached = counting try: - f1, s1 = gate_lib.run_gate(repo, {}) + f1, _s1 = gate_lib.run_gate(repo, {}) n_after_first = len(exec_count) - f2, s2 = gate_lib.run_gate(repo, {}) + f2, _s2 = gate_lib.run_gate(repo, {}) ok("门禁缓存命中(第二次零执行)", len(exec_count) == n_after_first == 1, f"执行 {len(exec_count)} 次") ok("缓存结果一致", [x[0] for x in f1] == [x[0] for x in f2]) @@ -560,7 +560,7 @@ def counting(root, cfg, langs): "gate_lib.run_gate=_boom\n" 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, text=True, + r = subprocess.run([sys.executable, "-c", code], capture_output=True, check=False, text=True, cwd=PLUGIN, timeout=60) ok("钩子内部错误 fail-open(exit 0)", r.returncode == 0, f"exit={r.returncode}") ok("无 traceback 进输出", "Traceback" not in (r.stdout + r.stderr)) diff --git a/tests/test_gate_trigger.py b/tests/test_gate_trigger.py new file mode 100644 index 0000000..56bbd9f --- /dev/null +++ b/tests/test_gate_trigger.py @@ -0,0 +1,92 @@ +"""Group 1+2 regression tests: commit-gate trigger heuristics and targeted +language detection (OpenSpec change 2026-09-22-fix-gate-trigger-and-mcp). + +Grounded in the 2026-09-22 empirical audit: +- Real false positives are SUBSTRING matches: `pushed` hits `push`, + `deployment` hits `deploy`. +- ASCII `?` is already present in QUESTION_MARKERS; we lock it with a + regression test rather than "adding" it. +- Expected-True cases cover bare imperatives and action-intent pairings. +""" +from __future__ import annotations + +import importlib.util +import sys +import unittest +from pathlib import Path + +PLUGIN = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PLUGIN / "scripts")) +sys.path.insert(0, str(PLUGIN / "hooks")) + + +def _load_module(): + spec = importlib.util.spec_from_file_location( + "upv", PLUGIN / "hooks" / "user_prompt_validator.py" + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class TriggerHeuristicsTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.upv = _load_module() + + def test_substring_variants_do_not_trigger(self): + """1.1: word-boundary matching kills the real observed FPs.""" + for text in ("我刚才 pushed 了", "关于 deployment 策略的讨论"): + with self.subTest(text=text): + self.assertFalse(self.upv.is_trigger(text)) + + def test_question_markers_cover_both_languages(self): + """1.3: ASCII `?` and `?` both present (regression lock, not new).""" + markers = self.upv.QUESTION_MARKERS + self.assertIn("?", markers) + self.assertIn("?", markers) + # English question with trigger word is exempt. + self.assertFalse(self.upv.is_trigger("Should I commit this?")) + + def test_bare_imperative_still_triggers(self): + """1.4: sentence-initial triggers keep working.""" + for text in ("commit this now", "提交代码", "commit this"): + with self.subTest(text=text): + self.assertTrue(self.upv.is_trigger(text)) + + def test_action_intent_pairing_triggers(self): + """1.2: trigger word + action-intent phrase passes.""" + for text in ("请帮我 commit", "请帮我 push", "现在提交", "please push this"): + with self.subTest(text=text): + self.assertTrue(self.upv.is_trigger(text)) + + def test_no_action_intent_stays_silent(self): + """1.2: trigger word mid-sentence without action intent stays silent.""" + for text in ("我们聊聊 push 和 pull request 的区别"): + with self.subTest(text=text): + self.assertFalse(self.upv.is_trigger(text)) + + +class TargetedLanguageTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.upv = _load_module() + + def test_detects_languages_mentioned_in_text(self): + """2.1/2.2: prompt naming rust narrows the gate to rust.""" + if not hasattr(self.upv, "_detect_languages_in_text"): + self.skipTest("_detect_languages_in_text not implemented yet") + ids = ["java", "rust", "typescript", "python", "go"] + got = self.upv._detect_languages_in_text("commit this rust change please", ids) + self.assertEqual(got, ["rust"]) + + def test_falls_back_to_empty_when_no_language_named(self): + """2.3: no language token → empty subset → caller falls back to detect.""" + if not hasattr(self.upv, "_detect_languages_in_text"): + self.skipTest("_detect_languages_in_text not implemented yet") + got = self.upv._detect_languages_in_text("请帮我 commit", ["java", "rust"]) + self.assertEqual(got, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py new file mode 100644 index 0000000..ea4a524 --- /dev/null +++ b/tests/test_mcp_server.py @@ -0,0 +1,196 @@ +"""Groups 3+4 regression tests: MCP server via official SDK + failure log +(OpenSpec change 2026-09-22-fix-gate-trigger-and-mcp). + +MCP tests skip gracefully when the `mcp` SDK is not installed (CI installs +requirements.txt; local dev may not). +""" +from __future__ import annotations + +import importlib.util +import json +import selectors +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +PLUGIN = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PLUGIN / "scripts")) + +MCP_AVAILABLE = importlib.util.find_spec("mcp") is not None +RUN_CHECK = PLUGIN / "scripts" / "run_check.py" + + +def _spawn_mcp() -> subprocess.Popen: + return subprocess.Popen( + [sys.executable, str(RUN_CHECK), "--mcp", str(PLUGIN)], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, bufsize=1, + ) + + +def _send(proc: subprocess.Popen, msg: dict) -> None: + assert proc.stdin is not None + proc.stdin.write(json.dumps(msg) + "\n") + proc.stdin.flush() + + +def _read_until(proc: subprocess.Popen, want_id: int, timeout: float = 30.0) -> dict: + assert proc.stdout is not None + sel = selectors.DefaultSelector() + sel.register(proc.stdout, selectors.EVENT_READ) + deadline_line = None + import time + end = time.monotonic() + timeout + while time.monotonic() < end: + if not sel.select(timeout=1.0): + continue + line = proc.stdout.readline() + if not line: + break + line = line.strip() + if not line: + continue + try: + msg = json.loads(line) + except json.JSONDecodeError: + continue + if msg.get("id") == want_id: + deadline_line = msg + break + sel.close() + if deadline_line is None: + err = proc.stderr.read() if proc.stderr else "" + raise AssertionError(f"no response for id={want_id}; stderr={err[:2000]!r}") + return deadline_line + + +@unittest.skipUnless(MCP_AVAILABLE, "mcp SDK not installed (pip install -r requirements.txt)") +class McpServerTests(unittest.TestCase): + def test_mcp_boot_lists_three_tools(self): + """3.2/6.3: server boots over stdio and exposes exactly the 3 tools.""" + proc = _spawn_mcp() + try: + _send(proc, { + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "0"}, + }, + }) + init = _read_until(proc, 1) + self.assertIn("result", init, init) + _send(proc, {"jsonrpc": "2.0", "method": "notifications/initialized"}) + _send(proc, {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) + listing = _read_until(proc, 2) + names = {t["name"] for t in listing["result"]["tools"]} + self.assertEqual(names, {"check_code_style", "auto_fix", "list_languages"}) + finally: + try: + if proc.stdin: + proc.stdin.close() + proc.wait(timeout=10) + except Exception: # noqa: BLE001 cleanup + proc.kill() + + def test_mcp_list_languages_returns_id_and_name_only(self): + """3.5/6.3: tool result enumerates id+name, never internal commands.""" + proc = _spawn_mcp() + try: + _send(proc, { + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "0"}, + }, + }) + _read_until(proc, 1) + _send(proc, {"jsonrpc": "2.0", "method": "notifications/initialized"}) + _send(proc, { + "jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "list_languages", "arguments": {}}, + }) + result = _read_until(proc, 3) + content = result["result"]["content"] + payload = json.loads(content[0]["text"]) + self.assertIsInstance(payload, list) + self.assertGreater(len(payload), 10) + for entry in payload: + self.assertEqual(set(entry.keys()), {"id", "name"}) + finally: + try: + if proc.stdin: + proc.stdin.close() + proc.wait(timeout=10) + except Exception: # noqa: BLE001 cleanup + proc.kill() + + +class FailureLogTests(unittest.TestCase): + def _bad_python_project(self) -> tuple[tempfile.TemporaryDirectory, Path]: + tmp = tempfile.TemporaryDirectory() + root = Path(tmp.name) + (root / "bad.py").write_text("import os\n", encoding="utf-8") + return tmp, root + + def test_run_check_writes_full_stderr_log_on_failure(self): + """4.1/4.2: failing lint writes /.codeguard-last.log and + attaches log_path to the failing result entry.""" + import run_per_language + tmp, root = self._bad_python_project() + try: + log_dir = root / "out" + results = run_per_language.run_check(["python"], root, log_dir=log_dir) + self.assertEqual(len(results), 1) + r = results[0] + # ruff present → F401 failure; ruff absent → 127 command not found. + # Either way passed=False with a real log written. + self.assertFalse(r["passed"]) + log_path = Path(r["log_path"]) + self.assertTrue(log_path.is_file(), "log file must exist") + content = log_path.read_text(encoding="utf-8") + self.assertIn("python", content) + self.assertTrue(content.strip(), "log must contain stderr content") + finally: + tmp.cleanup() + + def test_cli_failure_prints_log_path(self): + """4.2 (CLI): terminal gets a summary line + absolute log path.""" + tmp, root = self._bad_python_project() + try: + log_dir = root / "out" + proc = subprocess.run( + [sys.executable, str(RUN_CHECK), str(root), + "--lang", "python", "--log-dir", str(log_dir)], + capture_output=True, check=False, text=True, timeout=120, + ) + self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr) + self.assertIn("完整日志", proc.stdout) + expected_log = (log_dir / ".codeguard-last.log").resolve() + self.assertIn(str(expected_log), proc.stdout) + self.assertTrue(expected_log.is_file()) + finally: + tmp.cleanup() + + def test_cli_quiet_suppresses_log_and_summary(self): + """4.3: --quiet writes no log and prints no log summary line.""" + tmp, root = self._bad_python_project() + try: + log_dir = root / "out" + proc = subprocess.run( + [sys.executable, str(RUN_CHECK), str(root), + "--lang", "python", "--log-dir", str(log_dir), "--quiet"], + capture_output=True, check=False, text=True, timeout=120, + ) + self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr) + self.assertNotIn("完整日志", proc.stdout) + self.assertFalse((log_dir / ".codeguard-last.log").exists()) + finally: + tmp.cleanup() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_readme_parity.py b/tests/test_readme_parity.py old mode 100644 new mode 100755 diff --git a/tests/test_skill_vendor.py b/tests/test_skill_vendor.py index d3c8214..80d469b 100644 --- a/tests/test_skill_vendor.py +++ b/tests/test_skill_vendor.py @@ -11,7 +11,6 @@ import unittest from pathlib import Path - ROOT = Path(__file__).resolve().parents[1] SCRIPT = ROOT / "scripts" / "vendor" / "skill_vendor.py" GIT_ENV = { @@ -83,7 +82,7 @@ def vendor(command: str, consumer: Path, *extra: str) -> subprocess.CompletedPro return subprocess.run( [sys.executable, str(SCRIPT), command, "--lock", "skills.lock.json", *extra], cwd=consumer, - capture_output=True, + capture_output=True, check=False, text=True, ) diff --git a/tests/test_validate_languages_json.py b/tests/test_validate_languages_json.py old mode 100644 new mode 100755