wehub-resource-sync
1482 lines
55 KiB
Python
1482 lines
55 KiB
Python
"""Claude Code skills and hooks auto-install.
|
||
|
||
Generates Claude Code agent skill files, hooks configuration, and
|
||
CLAUDE.md integration for seamless code-review-graph usage.
|
||
Also supports multi-platform MCP server installation and
|
||
Cursor hooks / OpenCode plugin generation.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import platform
|
||
import re
|
||
import shutil
|
||
import stat
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# --- Multi-platform MCP install ---
|
||
|
||
|
||
def _zed_settings_path() -> Path:
|
||
"""Return the Zed settings.json path for the current OS."""
|
||
if platform.system() == "Darwin":
|
||
return Path.home() / "Library" / "Application Support" / "Zed" / "settings.json"
|
||
return Path.home() / ".config" / "zed" / "settings.json"
|
||
|
||
|
||
PLATFORMS: dict[str, dict[str, Any]] = {
|
||
"codex": {
|
||
"name": "Codex",
|
||
"config_path": lambda root: Path.home() / ".codex" / "config.toml",
|
||
"key": "mcp_servers",
|
||
"detect": lambda: (Path.home() / ".codex").exists(),
|
||
"format": "toml",
|
||
"needs_type": True,
|
||
},
|
||
"claude": {
|
||
"name": "Claude Code",
|
||
"config_path": lambda root: root / ".mcp.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: True,
|
||
"format": "object",
|
||
"needs_type": True,
|
||
},
|
||
"cursor": {
|
||
"name": "Cursor",
|
||
"config_path": lambda root: root / ".cursor" / "mcp.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: (Path.home() / ".cursor").exists(),
|
||
"format": "object",
|
||
"needs_type": True,
|
||
},
|
||
"windsurf": {
|
||
"name": "Windsurf",
|
||
"config_path": lambda root: Path.home() / ".codeium" / "windsurf" / "mcp_config.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: (Path.home() / ".codeium" / "windsurf").exists(),
|
||
"format": "object",
|
||
"needs_type": False,
|
||
},
|
||
"zed": {
|
||
"name": "Zed",
|
||
"config_path": lambda root: _zed_settings_path(),
|
||
"key": "context_servers",
|
||
"detect": lambda: _zed_settings_path().parent.exists(),
|
||
"format": "object",
|
||
"needs_type": False,
|
||
},
|
||
"continue": {
|
||
"name": "Continue",
|
||
"config_path": lambda root: Path.home() / ".continue" / "config.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: (Path.home() / ".continue").exists(),
|
||
"format": "array",
|
||
"needs_type": True,
|
||
},
|
||
"opencode": {
|
||
"name": "OpenCode",
|
||
"config_path": lambda root: root / ".opencode.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: True,
|
||
"format": "object",
|
||
"needs_type": True,
|
||
},
|
||
"antigravity": {
|
||
"name": "Antigravity",
|
||
"config_path": lambda root: Path.home() / ".gemini" / "antigravity" / "mcp_config.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: (Path.home() / ".gemini" / "antigravity").exists(),
|
||
"format": "object",
|
||
"needs_type": False,
|
||
},
|
||
"gemini-cli": {
|
||
"name": "Gemini CLI",
|
||
"config_path": lambda root: root / ".gemini" / "settings.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: bool(shutil.which("gemini")) or (Path.home() / ".gemini").exists(),
|
||
"format": "object",
|
||
"needs_type": False,
|
||
},
|
||
"qwen": {
|
||
"name": "Qwen Code",
|
||
"config_path": lambda root: Path.home() / ".qwen" / "settings.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: (Path.home() / ".qwen").exists(),
|
||
"format": "object",
|
||
"needs_type": True,
|
||
},
|
||
"kiro": {
|
||
"name": "Kiro",
|
||
"config_path": lambda root: root / ".kiro" / "settings" / "mcp.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: (Path.home() / ".kiro").exists(),
|
||
"format": "object",
|
||
"needs_type": True,
|
||
},
|
||
"qoder": {
|
||
"name": "Qoder",
|
||
"config_path": lambda root: root / ".qoder" / "mcp.json",
|
||
"key": "mcpServers",
|
||
"detect": lambda: True,
|
||
"format": "object",
|
||
"needs_type": True,
|
||
},
|
||
"copilot": {
|
||
"name": "GitHub Copilot",
|
||
"config_path": lambda root: root / ".vscode" / "mcp.json",
|
||
"key": "servers",
|
||
"detect": lambda: (Path.home() / ".vscode").exists(),
|
||
"format": "object",
|
||
"needs_type": True,
|
||
},
|
||
"copilot-cli": {
|
||
"name": "GitHub Copilot CLI",
|
||
"config_path": lambda root: Path.home() / ".copilot" / "mcp-config.json",
|
||
"key": "servers",
|
||
"detect": lambda: (Path.home() / ".copilot").exists(),
|
||
"format": "object",
|
||
"needs_type": True,
|
||
},
|
||
}
|
||
|
||
|
||
def _in_poetry_project() -> bool:
|
||
"""Return True when the running interpreter is a Poetry-managed virtualenv.
|
||
|
||
Two signals are checked so that **both** ``poetry shell`` and ``poetry run``
|
||
are detected:
|
||
|
||
* ``POETRY_ACTIVE=1`` — set by ``poetry shell`` when the user activates the
|
||
virtual environment interactively.
|
||
* ``VIRTUAL_ENV`` containing ``"pypoetry"`` — set by **both** ``poetry shell``
|
||
and ``poetry run`` because Poetry stores its virtualenvs under a path that
|
||
includes the string ``pypoetry`` (e.g.
|
||
``~/.cache/pypoetry/virtualenvs/<name>`` on Linux/macOS or
|
||
``%LOCALAPPDATA%\\pypoetry\\Cache\\virtualenvs\\<name>`` on Windows).
|
||
|
||
Checking only ``POETRY_ACTIVE`` would miss the ``poetry run`` case, which is
|
||
the primary scenario described in issue #256.
|
||
"""
|
||
if os.environ.get("POETRY_ACTIVE") == "1":
|
||
return True
|
||
virtual_env = os.environ.get("VIRTUAL_ENV", "")
|
||
return bool(virtual_env) and "pypoetry" in virtual_env.lower()
|
||
|
||
|
||
def _in_uv_project() -> bool:
|
||
"""Return True if ``sys.executable`` lives inside a uv-managed project.
|
||
|
||
A project is considered uv-managed when a ``uv.lock`` file exists in any
|
||
ancestor directory of the running Python interpreter (stopping at the home
|
||
directory to avoid false positives on system-wide installations).
|
||
"""
|
||
exe = Path(sys.executable).resolve()
|
||
home = Path.home()
|
||
for parent in exe.parents:
|
||
if (parent / "uv.lock").exists():
|
||
return True
|
||
# Stop searching once we reach the home directory or filesystem root
|
||
if parent == home or parent == parent.parent:
|
||
break
|
||
return False
|
||
|
||
|
||
def _detect_serve_command() -> tuple[str, list[str]]:
|
||
"""Return ``(command, args)`` that correctly launches ``code-review-graph serve``.
|
||
|
||
Detection priority
|
||
------------------
|
||
1. **Poetry** – ``POETRY_ACTIVE=1`` OR ``VIRTUAL_ENV`` contains ``"pypoetry"``
|
||
(covers both ``poetry shell`` and ``poetry run``) and ``poetry`` is on PATH
|
||
→ ``poetry run code-review-graph serve``
|
||
2. **uv project** – ``UV_PROJECT_ENVIRONMENT`` is set, or a ``uv.lock``
|
||
ancestor is found alongside ``sys.executable``, and ``uv`` is on PATH
|
||
→ ``uv run code-review-graph serve``
|
||
3. **uvx** – ``uvx`` is available on PATH (existing behaviour, unchanged)
|
||
→ ``uvx code-review-graph serve``
|
||
4. **Fallback** – use the absolute path of the running Python interpreter
|
||
→ ``sys.executable -m code_review_graph serve``
|
||
|
||
The fallback is always safe: ``sys.executable`` is the exact interpreter
|
||
that is currently running, so it resolves correctly inside any virtual
|
||
environment, conda env, or system installation.
|
||
"""
|
||
# 1. Poetry (poetry shell or poetry run)
|
||
if _in_poetry_project():
|
||
poetry = shutil.which("poetry")
|
||
if poetry:
|
||
return ("poetry", ["run", "code-review-graph", "serve"])
|
||
|
||
# 2. uv managed project environment
|
||
if os.environ.get("UV_PROJECT_ENVIRONMENT") or _in_uv_project():
|
||
uv = shutil.which("uv")
|
||
if uv:
|
||
return ("uv", ["run", "code-review-graph", "serve"])
|
||
|
||
# 3. uvx global tool runner (existing behaviour, unchanged)
|
||
if shutil.which("uvx"):
|
||
return ("uvx", ["code-review-graph", "serve"])
|
||
|
||
# 4. Absolute-path fallback using the running interpreter
|
||
return (sys.executable, ["-m", "code_review_graph", "serve"])
|
||
|
||
|
||
def _build_server_entry(
|
||
plat: dict[str, Any], key: str = "", repo_root: "Path | None" = None,
|
||
) -> dict[str, Any]:
|
||
"""Build the MCP server entry for a platform."""
|
||
command, args = _detect_serve_command()
|
||
entry: dict[str, Any] = {"command": command, "args": args}
|
||
# Include cwd so the MCP server can find the graph database
|
||
if repo_root is not None:
|
||
entry["cwd"] = str(repo_root)
|
||
if plat["needs_type"]:
|
||
entry["type"] = "stdio"
|
||
if key == "opencode":
|
||
entry["env"] = []
|
||
return entry
|
||
|
||
|
||
def _format_toml_value(value: Any) -> str:
|
||
"""Format a primitive Python value as TOML."""
|
||
if isinstance(value, str):
|
||
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
|
||
return f'"{escaped}"'
|
||
if isinstance(value, bool):
|
||
return "true" if value else "false"
|
||
if isinstance(value, list):
|
||
return "[" + ", ".join(_format_toml_value(item) for item in value) + "]"
|
||
raise TypeError(f"Unsupported TOML value: {type(value)!r}")
|
||
|
||
|
||
def _merge_toml_mcp_server(
|
||
config_path: Path,
|
||
server_name: str,
|
||
server_entry: dict[str, Any],
|
||
dry_run: bool = False,
|
||
) -> bool:
|
||
"""Append a Codex MCP server section without clobbering the rest of the file."""
|
||
section_header = f"[mcp_servers.{server_name}]"
|
||
existing = ""
|
||
if config_path.exists():
|
||
existing = config_path.read_text(encoding="utf-8")
|
||
if section_header in existing:
|
||
return False
|
||
|
||
section_lines = [section_header]
|
||
for key, value in server_entry.items():
|
||
section_lines.append(f"{key} = {_format_toml_value(value)}")
|
||
section = "\n".join(section_lines) + "\n"
|
||
|
||
if dry_run:
|
||
return True
|
||
|
||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||
prefix = ""
|
||
if existing:
|
||
prefix = existing if existing.endswith("\n") else existing + "\n"
|
||
if not prefix.endswith("\n\n"):
|
||
prefix += "\n"
|
||
config_path.write_text(prefix + section, encoding="utf-8")
|
||
return True
|
||
|
||
|
||
def install_platform_configs(
|
||
repo_root: Path,
|
||
target: str = "all",
|
||
dry_run: bool = False,
|
||
) -> list[str]:
|
||
"""Install MCP config for one or all detected platforms.
|
||
|
||
Args:
|
||
repo_root: Project root directory.
|
||
target: Platform key or "all".
|
||
dry_run: If True, print what would be done without writing.
|
||
|
||
Returns:
|
||
List of platform names that were configured.
|
||
"""
|
||
if target == "all":
|
||
platforms_to_install = {k: v for k, v in PLATFORMS.items() if v["detect"]()}
|
||
# Workspace-level Kiro detection
|
||
if "kiro" not in platforms_to_install and (repo_root / ".kiro").is_dir():
|
||
platforms_to_install["kiro"] = PLATFORMS["kiro"]
|
||
else:
|
||
if target not in PLATFORMS:
|
||
logger.error("Unknown platform: %s", target)
|
||
return []
|
||
platforms_to_install = {target: PLATFORMS[target]}
|
||
|
||
configured: list[str] = []
|
||
|
||
for key, plat in platforms_to_install.items():
|
||
config_path: Path = plat["config_path"](repo_root)
|
||
server_key = plat["key"]
|
||
server_entry = _build_server_entry(plat, key=key, repo_root=repo_root)
|
||
|
||
if plat["format"] == "toml":
|
||
changed = _merge_toml_mcp_server(
|
||
config_path,
|
||
"code-review-graph",
|
||
server_entry,
|
||
dry_run=dry_run,
|
||
)
|
||
if not changed:
|
||
print(f" {plat['name']}: already configured in {config_path}")
|
||
configured.append(plat["name"])
|
||
continue
|
||
if dry_run:
|
||
print(f" [dry-run] {plat['name']}: would write {config_path}")
|
||
else:
|
||
print(f" {plat['name']}: configured {config_path}")
|
||
configured.append(plat["name"])
|
||
continue
|
||
|
||
# Read existing config
|
||
existing: dict[str, Any] = {}
|
||
if config_path.exists():
|
||
raw = config_path.read_text(encoding="utf-8", errors="replace")
|
||
# Strip single-line comments and trailing commas (JSONC compat
|
||
# for editors like Zed that allow non-standard JSON).
|
||
stripped = re.sub(r'//.*?$', '', raw, flags=re.MULTILINE)
|
||
stripped = re.sub(r',(\s*[}\]])', r'\1', stripped)
|
||
try:
|
||
existing = json.loads(stripped)
|
||
except (json.JSONDecodeError, OSError):
|
||
print(f" {plat['name']}: {config_path} contains "
|
||
f"unparseable JSON — skipping to avoid data loss. "
|
||
f"Please add the MCP config manually.")
|
||
continue
|
||
|
||
if plat["format"] == "array":
|
||
arr = existing.get(server_key, [])
|
||
if not isinstance(arr, list):
|
||
arr = []
|
||
# Check if already present
|
||
if any(isinstance(s, dict) and s.get("name") == "code-review-graph" for s in arr):
|
||
print(f" {plat['name']}: already configured in {config_path}")
|
||
configured.append(plat["name"])
|
||
continue
|
||
arr_entry = {"name": "code-review-graph", **server_entry}
|
||
arr.append(arr_entry)
|
||
existing[server_key] = arr
|
||
else:
|
||
servers = existing.get(server_key, {})
|
||
if not isinstance(servers, dict):
|
||
servers = {}
|
||
if "code-review-graph" in servers:
|
||
print(f" {plat['name']}: already configured in {config_path}")
|
||
configured.append(plat["name"])
|
||
continue
|
||
servers["code-review-graph"] = server_entry
|
||
existing[server_key] = servers
|
||
|
||
if dry_run:
|
||
print(f" [dry-run] {plat['name']}: would write {config_path}")
|
||
else:
|
||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||
config_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
|
||
print(f" {plat['name']}: configured {config_path}")
|
||
|
||
configured.append(plat["name"])
|
||
|
||
return configured
|
||
|
||
|
||
# --- Skill file contents ---
|
||
|
||
_SKILLS: dict[str, dict[str, str]] = {
|
||
"explore-codebase.md": {
|
||
"name": "Explore Codebase",
|
||
"description": "Navigate and understand codebase structure using the knowledge graph",
|
||
"body": (
|
||
"## Explore Codebase\n\n"
|
||
"Use the code-review-graph MCP tools to explore and understand the codebase.\n\n"
|
||
"### Steps\n\n"
|
||
"1. Run `list_graph_stats` to see overall codebase metrics.\n"
|
||
"2. Run `get_architecture_overview_tool` for high-level community structure.\n"
|
||
"3. Use `list_communities_tool` to find major modules, then `get_community` "
|
||
"for details.\n"
|
||
"4. Use `semantic_search_nodes_tool` to find specific functions or classes.\n"
|
||
"5. Use `query_graph_tool` with patterns like `callers_of`, `callees_of`, "
|
||
"`imports_of` to trace relationships.\n"
|
||
"6. Use `list_flows` and `get_flow` to understand execution paths.\n\n"
|
||
"### Tips\n\n"
|
||
"- Start broad (stats, architecture) then narrow down to specific areas.\n"
|
||
"- Use `children_of` on a file to see all its functions and classes.\n"
|
||
"- Use `find_large_functions` to identify complex code.\n\n"
|
||
"## Token Efficiency Rules\n"
|
||
'- ALWAYS start with `get_minimal_context(task="<your task>")` '
|
||
"before any other graph tool.\n"
|
||
'- Use `detail_level="minimal"` on all calls. Only escalate to '
|
||
'"standard" when minimal is insufficient.\n'
|
||
"- Target: complete any review/debug/refactor task in ≤5 tool calls "
|
||
"and ≤800 total output tokens."
|
||
),
|
||
},
|
||
"review-changes.md": {
|
||
"name": "Review Changes",
|
||
"description": "Perform a structured code review using change detection and impact",
|
||
"body": (
|
||
"## Review Changes\n\n"
|
||
"Perform a thorough, risk-aware code review using the knowledge graph.\n\n"
|
||
"### Steps\n\n"
|
||
"1. Run `detect_changes_tool` to get risk-scored change analysis.\n"
|
||
"2. Run `get_affected_flows_tool` to find impacted execution paths.\n"
|
||
"3. For each high-risk function, run `query_graph_tool` with "
|
||
'pattern="tests_for" to check test coverage.\n'
|
||
"4. Run `get_impact_radius_tool` to understand the blast radius.\n"
|
||
"5. For any untested changes, suggest specific test cases.\n\n"
|
||
"### Output Format\n\n"
|
||
"Provide findings grouped by risk level (high/medium/low) with:\n"
|
||
"- What changed and why it matters\n"
|
||
"- Test coverage status\n"
|
||
"- Suggested improvements\n"
|
||
"- Overall merge recommendation\n\n"
|
||
"## Token Efficiency Rules\n"
|
||
'- ALWAYS start with `get_minimal_context(task="<your task>")` '
|
||
"before any other graph tool.\n"
|
||
'- Use `detail_level="minimal"` on all calls. Only escalate to '
|
||
'"standard" when minimal is insufficient.\n'
|
||
"- Target: complete any review/debug/refactor task in ≤5 tool calls "
|
||
"and ≤800 total output tokens."
|
||
),
|
||
},
|
||
"debug-issue.md": {
|
||
"name": "Debug Issue",
|
||
"description": "Systematically debug issues using graph-powered code navigation",
|
||
"body": (
|
||
"## Debug Issue\n\n"
|
||
"Use the knowledge graph to systematically trace and debug issues.\n\n"
|
||
"### Steps\n\n"
|
||
"1. Use `semantic_search_nodes_tool` to find code related to the issue.\n"
|
||
"2. Use `query_graph_tool` with `callers_of` and `callees_of` to trace "
|
||
"call chains.\n"
|
||
"3. Use `get_flow` to see full execution paths through suspected areas.\n"
|
||
"4. Run `detect_changes_tool` to check if recent changes caused the issue.\n"
|
||
"5. Use `get_impact_radius_tool` on suspected files to see what else is affected.\n\n"
|
||
"### Tips\n\n"
|
||
"- Check both callers and callees to understand the full context.\n"
|
||
"- Look at affected flows to find the entry point that triggers the bug.\n"
|
||
"- Recent changes are the most common source of new issues.\n\n"
|
||
"## Token Efficiency Rules\n"
|
||
'- ALWAYS start with `get_minimal_context(task="<your task>")` '
|
||
"before any other graph tool.\n"
|
||
'- Use `detail_level="minimal"` on all calls. Only escalate to '
|
||
'"standard" when minimal is insufficient.\n'
|
||
"- Target: complete any review/debug/refactor task in ≤5 tool calls "
|
||
"and ≤800 total output tokens."
|
||
),
|
||
},
|
||
"refactor-safely.md": {
|
||
"name": "Refactor Safely",
|
||
"description": "Plan and execute safe refactoring using dependency analysis",
|
||
"body": (
|
||
"## Refactor Safely\n\n"
|
||
"Use the knowledge graph to plan and execute refactoring with confidence.\n\n"
|
||
"### Steps\n\n"
|
||
'1. Use `refactor_tool` with mode="suggest" for community-driven '
|
||
"refactoring suggestions.\n"
|
||
'2. Use `refactor_tool` with mode="dead_code" to find unreferenced code.\n'
|
||
'3. For renames, use `refactor_tool` with mode="rename" to preview all '
|
||
"affected locations.\n"
|
||
"4. Use `apply_refactor_tool` with the refactor_id to apply renames.\n"
|
||
"5. After changes, run `detect_changes_tool` to verify the refactoring impact.\n\n"
|
||
"### Safety Checks\n\n"
|
||
"- Always preview before applying (rename mode gives you an edit list).\n"
|
||
"- Check `get_impact_radius_tool` before major refactors.\n"
|
||
"- Use `get_affected_flows_tool` to ensure no critical paths are broken.\n"
|
||
"- Run `find_large_functions` to identify decomposition targets.\n\n"
|
||
"## Token Efficiency Rules\n"
|
||
'- ALWAYS start with `get_minimal_context(task="<your task>")` '
|
||
"before any other graph tool.\n"
|
||
'- Use `detail_level="minimal"` on all calls. Only escalate to '
|
||
'"standard" when minimal is insufficient.\n'
|
||
"- Target: complete any review/debug/refactor task in ≤5 tool calls "
|
||
"and ≤800 total output tokens."
|
||
),
|
||
},
|
||
}
|
||
|
||
|
||
def generate_skills(repo_root: Path, skills_dir: Path | None = None) -> Path:
|
||
"""Generate Claude Code skill files.
|
||
|
||
Creates `.claude/skills/` directory with 4 skill markdown files,
|
||
each containing frontmatter and instructions.
|
||
|
||
Args:
|
||
repo_root: Repository root directory.
|
||
skills_dir: Custom skills directory. Defaults to repo_root/.claude/skills.
|
||
|
||
Returns:
|
||
Path to the skills directory.
|
||
"""
|
||
if skills_dir is None:
|
||
skills_dir = repo_root / ".claude" / "skills"
|
||
skills_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
for filename, skill in _SKILLS.items():
|
||
# Claude Code expects skills at .claude/skills/<name>/skill.md
|
||
skill_name = filename.removesuffix(".md")
|
||
skill_subdir = skills_dir / skill_name
|
||
skill_subdir.mkdir(parents=True, exist_ok=True)
|
||
path = skill_subdir / "skill.md"
|
||
content = (
|
||
"---\n"
|
||
f"name: {skill['name']}\n"
|
||
f"description: {skill['description']}\n"
|
||
"---\n\n"
|
||
f"{skill['body']}\n"
|
||
)
|
||
path.write_text(content, encoding="utf-8")
|
||
logger.info("Wrote skill: %s", path)
|
||
|
||
return skills_dir
|
||
|
||
|
||
def generate_hooks_config(repo_root: Path) -> dict[str, Any]:
|
||
"""Generate Claude Code hooks configuration.
|
||
|
||
Hooks use the v1.x+ schema: each entry needs a ``matcher`` and a nested
|
||
``hooks`` array. Timeouts are in seconds. ``PreCommit`` is not a valid
|
||
Claude Code event — pre-commit checks are handled by ``install_git_hook``.
|
||
"""
|
||
repo_arg = json.dumps(repo_root.resolve().as_posix())
|
||
return {
|
||
"hooks": {
|
||
"PostToolUse": [
|
||
{
|
||
"matcher": "Edit|Write|Bash",
|
||
"hooks": [
|
||
{
|
||
"type": "command",
|
||
"command": (
|
||
"cat >/dev/null || true; "
|
||
"git rev-parse --git-dir >/dev/null 2>&1"
|
||
f" && code-review-graph update --skip-flows"
|
||
f" --repo {repo_arg}"
|
||
" || true"
|
||
),
|
||
"timeout": 30,
|
||
},
|
||
],
|
||
},
|
||
],
|
||
"SessionStart": [
|
||
{
|
||
"matcher": "",
|
||
"hooks": [
|
||
{
|
||
"type": "command",
|
||
"command": (
|
||
"cat >/dev/null || true; "
|
||
"git rev-parse --git-dir >/dev/null 2>&1"
|
||
f" && code-review-graph status --repo {repo_arg}"
|
||
" || echo 'Not a git repo, skipping'"
|
||
),
|
||
"timeout": 10,
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
}
|
||
|
||
|
||
def generate_codex_hooks_config(repo_root: Path) -> dict[str, Any]:
|
||
"""Generate native Codex hooks configuration for ~/.codex/hooks.json."""
|
||
return {
|
||
"hooks": {
|
||
"PostToolUse": [
|
||
{
|
||
"matcher": "Write|Edit|Bash",
|
||
"hooks": [
|
||
{
|
||
"type": "command",
|
||
"command": (
|
||
"cat >/dev/null || true; "
|
||
"git rev-parse --git-dir >/dev/null 2>&1"
|
||
" && code-review-graph update --skip-flows"
|
||
" || true"
|
||
),
|
||
"timeout": 30,
|
||
"statusMessage": "Updating code-review-graph",
|
||
},
|
||
],
|
||
},
|
||
],
|
||
"SessionStart": [
|
||
{
|
||
"matcher": "startup|resume",
|
||
"hooks": [
|
||
{
|
||
"type": "command",
|
||
"command": (
|
||
"cat >/dev/null || true; "
|
||
"git rev-parse --git-dir >/dev/null 2>&1"
|
||
" && code-review-graph status"
|
||
" || echo 'Not a git repo, skipping'"
|
||
),
|
||
"timeout": 10,
|
||
"statusMessage": "Checking code-review-graph status",
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
}
|
||
|
||
|
||
def install_git_hook(repo_root: Path) -> Path | None:
|
||
"""Install a git pre-commit hook that prints a risk summary before each commit.
|
||
|
||
Called automatically by ``code-review-graph install``.
|
||
The hooks directory is resolved via ``git rev-parse --git-path hooks`` so
|
||
the hook lands where git actually runs it — including linked worktrees
|
||
and submodules (where ``.git`` is a file, not a directory) and repos with
|
||
``core.hooksPath`` set (issue #313). ``core.hooksPath`` users with their
|
||
own hook manager (husky, pre-commit) may prefer integrating the
|
||
``code-review-graph`` commands into that manager manually instead.
|
||
|
||
Creates ``pre-commit`` if it doesn't exist, or appends to an existing
|
||
one — the hook is appended, not overwritten, preserving any hooks
|
||
already there. Falls back to the legacy ``.git/hooks`` resolution when
|
||
git itself is unavailable. Returns None when no hooks directory can be
|
||
determined.
|
||
"""
|
||
script = """\
|
||
#!/bin/sh
|
||
# Installed by code-review-graph. Remove this file to disable pre-commit graph checks.
|
||
if command -v code-review-graph >/dev/null 2>&1; then
|
||
code-review-graph update || true
|
||
code-review-graph detect-changes --brief || true
|
||
fi
|
||
"""
|
||
marker = "code-review-graph detect-changes"
|
||
|
||
hooks_dir: Path | None = None
|
||
try:
|
||
result = subprocess.run(
|
||
["git", "rev-parse", "--git-path", "hooks"],
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
cwd=str(repo_root),
|
||
timeout=10,
|
||
stdin=subprocess.DEVNULL,
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
# Output is relative to repo_root (".git/hooks", a core.hooksPath
|
||
# value such as ".husky") or absolute (linked worktrees).
|
||
hooks_dir = repo_root / result.stdout.strip()
|
||
except (subprocess.TimeoutExpired, OSError) as exc:
|
||
logger.warning("git unavailable (%s); falling back to .git/hooks resolution.", exc)
|
||
|
||
if hooks_dir is None:
|
||
git_dir = repo_root / ".git"
|
||
if not git_dir.is_dir():
|
||
logger.warning(
|
||
"No git hooks directory found at %s — skipping git hook install.", repo_root
|
||
)
|
||
return None
|
||
hooks_dir = git_dir / "hooks"
|
||
|
||
hook_path = hooks_dir / "pre-commit"
|
||
hook_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
if hook_path.exists():
|
||
existing = hook_path.read_text(encoding="utf-8")
|
||
if marker in existing:
|
||
return hook_path
|
||
hook_path.write_text(existing.rstrip("\n") + "\n" + script, encoding="utf-8")
|
||
else:
|
||
hook_path.write_text(script, encoding="utf-8")
|
||
|
||
hook_path.chmod(0o755)
|
||
logger.info("Wrote git pre-commit hook: %s", hook_path)
|
||
return hook_path
|
||
|
||
|
||
def install_hooks(repo_root: Path, platform: str = "claude") -> None:
|
||
"""Write hooks config to platform-specific settings.json.
|
||
|
||
Merges new hook entries into existing settings, preserving both
|
||
non-hook configuration and user-defined hooks. A backup of the
|
||
original file is created before any modifications.
|
||
|
||
Args:
|
||
repo_root: Repository root directory.
|
||
platform: Target platform ("claude" or "qoder").
|
||
"""
|
||
if platform == "qoder":
|
||
settings_dir = repo_root / ".qoder"
|
||
else:
|
||
settings_dir = repo_root / ".claude"
|
||
settings_dir.mkdir(parents=True, exist_ok=True)
|
||
settings_path = settings_dir / "settings.json"
|
||
|
||
existing: dict[str, Any] = {}
|
||
if settings_path.exists():
|
||
try:
|
||
existing = json.loads(settings_path.read_text(encoding="utf-8", errors="replace"))
|
||
backup_path = settings_dir / "settings.json.bak"
|
||
shutil.copy2(settings_path, backup_path)
|
||
logger.info("Backed up existing settings to %s", backup_path)
|
||
except (json.JSONDecodeError, OSError) as exc:
|
||
logger.warning("Could not read existing %s: %s", settings_path, exc)
|
||
|
||
hooks_config = generate_hooks_config(repo_root)
|
||
existing_hooks = existing.get("hooks", {})
|
||
if not isinstance(existing_hooks, dict):
|
||
logger.warning("Existing hooks config is not a dict; replacing with defaults")
|
||
existing_hooks = {}
|
||
|
||
merged_hooks = dict(existing_hooks)
|
||
for hook_name, hook_entries in hooks_config.get("hooks", {}).items():
|
||
if isinstance(merged_hooks.get(hook_name), list):
|
||
merged_list = list(merged_hooks[hook_name])
|
||
for entry in hook_entries:
|
||
if entry not in merged_list:
|
||
merged_list.append(entry)
|
||
merged_hooks[hook_name] = merged_list
|
||
else:
|
||
merged_hooks[hook_name] = hook_entries
|
||
|
||
existing["hooks"] = merged_hooks
|
||
|
||
settings_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
|
||
logger.info("Wrote hooks config: %s", settings_path)
|
||
|
||
|
||
def install_codex_hooks(repo_root: Path) -> Path:
|
||
"""Write native Codex hooks config to ~/.codex/hooks.json.
|
||
|
||
Merges code-review-graph hook entries into any existing hooks.json,
|
||
preserving user-defined hook entries and other top-level settings.
|
||
A backup of the original file is created before modifications.
|
||
"""
|
||
codex_dir = Path.home() / ".codex"
|
||
codex_dir.mkdir(parents=True, exist_ok=True)
|
||
hooks_path = codex_dir / "hooks.json"
|
||
|
||
existing: dict[str, Any] = {}
|
||
if hooks_path.exists():
|
||
try:
|
||
existing = json.loads(hooks_path.read_text(encoding="utf-8", errors="replace"))
|
||
backup_path = codex_dir / "hooks.json.bak"
|
||
shutil.copy2(hooks_path, backup_path)
|
||
logger.info("Backed up existing Codex hooks to %s", backup_path)
|
||
except (json.JSONDecodeError, OSError) as exc:
|
||
logger.warning("Could not read existing %s: %s", hooks_path, exc)
|
||
|
||
hooks_config = generate_codex_hooks_config(repo_root)
|
||
existing_hooks = existing.get("hooks", {})
|
||
if not isinstance(existing_hooks, dict):
|
||
logger.warning("Existing Codex hooks config is not a dict; replacing with defaults")
|
||
existing_hooks = {}
|
||
|
||
merged_hooks = dict(existing_hooks)
|
||
for hook_name, hook_entries in hooks_config.get("hooks", {}).items():
|
||
if isinstance(merged_hooks.get(hook_name), list):
|
||
merged_list = list(merged_hooks[hook_name])
|
||
existing_commands = {
|
||
hook.get("command", "")
|
||
for entry in merged_list
|
||
if isinstance(entry, dict)
|
||
for hook in entry.get("hooks", [])
|
||
if isinstance(hook, dict)
|
||
}
|
||
for entry in hook_entries:
|
||
entry_commands = [
|
||
hook.get("command", "")
|
||
for hook in entry.get("hooks", [])
|
||
if isinstance(hook, dict)
|
||
]
|
||
if not any(command in existing_commands for command in entry_commands):
|
||
merged_list.append(entry)
|
||
merged_hooks[hook_name] = merged_list
|
||
else:
|
||
merged_hooks[hook_name] = hook_entries
|
||
|
||
existing["hooks"] = merged_hooks
|
||
hooks_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
|
||
logger.info("Wrote Codex hooks config: %s", hooks_path)
|
||
return hooks_path
|
||
|
||
|
||
_CLAUDE_MD_SECTION_MARKER = "<!-- code-review-graph MCP tools -->"
|
||
|
||
_CLAUDE_MD_SECTION = f"""{_CLAUDE_MD_SECTION_MARKER}
|
||
## MCP Tools: code-review-graph
|
||
|
||
**IMPORTANT: This project has a knowledge graph. ALWAYS use the
|
||
code-review-graph MCP tools BEFORE using Grep/Glob/Read to explore
|
||
the codebase.** The graph is faster, cheaper (fewer tokens), and gives
|
||
you structural context (callers, dependents, test coverage) that file
|
||
scanning cannot.
|
||
|
||
### When to use graph tools FIRST
|
||
|
||
- **Exploring code**: `semantic_search_nodes_tool` or `query_graph_tool` instead of Grep
|
||
- **Understanding impact**: `get_impact_radius_tool` instead of manually tracing imports
|
||
- **Code review**: `detect_changes_tool` + `get_review_context_tool` instead of reading entire files
|
||
- **Finding relationships**: `query_graph_tool` with callers_of/callees_of/imports_of/tests_for
|
||
- **Architecture questions**: `get_architecture_overview_tool` + `list_communities_tool`
|
||
|
||
Fall back to Grep/Glob/Read **only** when the graph doesn't cover what you need.
|
||
|
||
### Key Tools
|
||
|
||
| Tool | Use when |
|
||
| ------ | ---------- |
|
||
| `detect_changes_tool` | Reviewing code changes — gives risk-scored analysis |
|
||
| `get_review_context_tool` | Need source snippets for review — token-efficient |
|
||
| `get_impact_radius_tool` | Understanding blast radius of a change |
|
||
| `get_affected_flows_tool` | Finding which execution paths are impacted |
|
||
| `query_graph_tool` | Tracing callers, callees, imports, tests, dependencies |
|
||
| `semantic_search_nodes_tool` | Finding functions/classes by name or keyword |
|
||
| `get_architecture_overview_tool` | Understanding high-level codebase structure |
|
||
| `refactor_tool` | Planning renames, finding dead code |
|
||
|
||
### Workflow
|
||
|
||
1. The graph auto-updates on file changes (via hooks).
|
||
2. Use `detect_changes_tool` for code review.
|
||
3. Use `get_affected_flows_tool` to understand impact.
|
||
4. Use `query_graph_tool` pattern=\"tests_for\" to check coverage.
|
||
"""
|
||
|
||
# Copilot-specific instruction file content: uses VS Code tool references and
|
||
# includes YAML front matter so Copilot Chat applies it across the workspace.
|
||
_COPILOT_SECTION = f"""---
|
||
applyTo: '**'
|
||
description: >-
|
||
Use code-review-graph MCP tools for token-efficient
|
||
codebase exploration and code review.
|
||
---
|
||
|
||
{_CLAUDE_MD_SECTION_MARKER}
|
||
## MCP Tools: code-review-graph
|
||
|
||
**IMPORTANT: This project has a knowledge graph. ALWAYS use the
|
||
code-review-graph MCP tools BEFORE using file/search tools to
|
||
explore the codebase.** The graph is faster, cheaper (fewer
|
||
tokens), and gives you structural context (callers, dependents,
|
||
test coverage) that file scanning cannot.
|
||
|
||
### When to use graph tools FIRST
|
||
|
||
- **Exploring code**: `semantic_search_nodes_tool` or `query_graph_tool`
|
||
- **Understanding impact**: `get_impact_radius_tool`
|
||
- **Code review**: `detect_changes_tool` + `get_review_context_tool`
|
||
- **Finding relationships**: `query_graph_tool` callers_of/callees_of
|
||
- **Architecture questions**: `get_architecture_overview_tool`
|
||
|
||
Fall back to file/search tools **only** when the graph doesn't
|
||
cover what you need.
|
||
|
||
### Key Tools
|
||
|
||
| Tool | Use when |
|
||
| ------ | ---------- |
|
||
| `detect_changes_tool` | Risk-scored change analysis |
|
||
| `get_review_context_tool` | Token-efficient source snippets |
|
||
| `get_impact_radius_tool` | Blast radius of a change |
|
||
| `get_affected_flows_tool` | Impacted execution paths |
|
||
| `query_graph_tool` | Trace callers, callees, imports, tests |
|
||
| `semantic_search_nodes_tool` | Find functions/classes by keyword |
|
||
| `get_architecture_overview_tool` | High-level structure |
|
||
| `refactor_tool` | Rename planning, dead code |
|
||
|
||
### Workflow
|
||
|
||
1. The graph auto-updates on file changes (via hooks).
|
||
2. Use `detect_changes_tool` for code review.
|
||
3. Use `get_affected_flows_tool` to understand impact.
|
||
4. Use `query_graph_tool` pattern=\"tests_for\" to check coverage.
|
||
"""
|
||
|
||
# Maps instruction file path → (marker, section) for files that need content
|
||
# different from the default _CLAUDE_MD_SECTION.
|
||
_PLATFORM_INSTRUCTION_CUSTOM_SECTIONS: dict[str, tuple[str, str]] = {
|
||
".github/code-review-graph.instruction.md": (_CLAUDE_MD_SECTION_MARKER, _COPILOT_SECTION),
|
||
}
|
||
|
||
|
||
def _inject_instructions(file_path: Path, marker: str, section: str) -> bool:
|
||
"""Append an instruction section to a file if not already present.
|
||
|
||
Idempotent: checks if the marker is already present before appending.
|
||
Creates the file if it doesn't exist.
|
||
|
||
Returns True if the file was modified.
|
||
"""
|
||
existing = ""
|
||
if file_path.exists():
|
||
existing = file_path.read_text(encoding="utf-8", errors="replace")
|
||
|
||
if marker in existing:
|
||
logger.info("%s already contains instructions, skipping.", file_path.name)
|
||
return False
|
||
|
||
separator = "\n" if existing and not existing.endswith("\n") else ""
|
||
extra_newline = "\n" if existing else ""
|
||
file_path.parent.mkdir(parents=True, exist_ok=True)
|
||
file_path.write_text(existing + separator + extra_newline + section, encoding="utf-8")
|
||
logger.info("Appended MCP tools section to %s", file_path)
|
||
return True
|
||
|
||
|
||
def inject_claude_md(repo_root: Path) -> None:
|
||
"""Append MCP tools section to CLAUDE.md."""
|
||
_inject_instructions(
|
||
repo_root / "CLAUDE.md",
|
||
_CLAUDE_MD_SECTION_MARKER,
|
||
_CLAUDE_MD_SECTION,
|
||
)
|
||
|
||
|
||
# Cross-platform instruction files and which platforms own each one.
|
||
# Used to filter writes when the user passes --platform <X>: only files
|
||
# whose owner set includes the target (or "all") are written.
|
||
_PLATFORM_INSTRUCTION_FILES: dict[str, tuple[str, ...]] = {
|
||
"AGENTS.md": ("cursor", "opencode", "antigravity"),
|
||
"GEMINI.md": ("antigravity", "gemini-cli"),
|
||
".cursorrules": ("cursor",),
|
||
".windsurfrules": ("windsurf",),
|
||
"QODER.md": ("qoder",),
|
||
".kiro/steering/code-review-graph.md": ("kiro",),
|
||
".github/code-review-graph.instruction.md": ("copilot", "copilot-cli"),
|
||
}
|
||
|
||
|
||
# --- Gemini CLI hooks + skills (workspace-level: .gemini/) ---
|
||
|
||
|
||
def install_gemini_cli_hooks(repo_root: Path) -> Path:
|
||
"""Install Gemini CLI hooks in .gemini/settings.json and write hook scripts.
|
||
|
||
Hooks schema reference:
|
||
- https://geminicli.com/docs/hooks/reference/
|
||
|
||
This is workspace-scoped (project) configuration: .gemini/settings.json
|
||
"""
|
||
settings_dir = repo_root / ".gemini"
|
||
settings_dir.mkdir(parents=True, exist_ok=True)
|
||
settings_path = settings_dir / "settings.json"
|
||
|
||
existing: dict[str, Any] = {}
|
||
if settings_path.exists():
|
||
try:
|
||
existing = json.loads(settings_path.read_text(encoding="utf-8", errors="replace"))
|
||
backup_path = settings_dir / "settings.json.bak"
|
||
shutil.copy2(settings_path, backup_path)
|
||
logger.info("Backed up existing Gemini CLI settings to %s", backup_path)
|
||
except (json.JSONDecodeError, OSError) as exc:
|
||
logger.warning("Could not read existing %s: %s", settings_path, exc)
|
||
|
||
hooks_dir = settings_dir / "hooks"
|
||
hooks_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
repo_arg = repo_root.resolve().as_posix()
|
||
session_start_script = """\
|
||
#!/usr/bin/env bash
|
||
# code-review-graph: session start status (Gemini CLI hook)
|
||
# Must output ONLY JSON on stdout. Logs go to stderr. Never blocks the session.
|
||
set -euo pipefail
|
||
|
||
cat > /dev/null || true
|
||
|
||
msg="$(code-review-graph status --repo "__CRG_REPO__" 2>&1 | head -n 1 || true)"
|
||
|
||
CRG_MSG="$msg" python3 -c '
|
||
import json,os
|
||
m=os.environ.get("CRG_MSG","")
|
||
print(json.dumps({"systemMessage":m,"suppressOutput":True}))
|
||
' 2>/dev/null || echo '{"suppressOutput": true}'
|
||
exit 0
|
||
"""
|
||
session_start_script = session_start_script.replace("__CRG_REPO__", repo_arg)
|
||
|
||
update_script = """\
|
||
#!/usr/bin/env bash
|
||
# code-review-graph: incremental update after write/replace (Gemini CLI hook)
|
||
# Must output ONLY JSON on stdout. Low-noise: no systemMessage.
|
||
set -euo pipefail
|
||
|
||
cat > /dev/null || true
|
||
|
||
code-review-graph update --skip-flows --repo "__CRG_REPO__" >/dev/null 2>&1 || true
|
||
echo '{"suppressOutput": true}'
|
||
exit 0
|
||
"""
|
||
update_script = update_script.replace("__CRG_REPO__", repo_arg)
|
||
|
||
session_start_path = hooks_dir / "crg-session-start.sh"
|
||
session_start_path.write_text(session_start_script, encoding="utf-8")
|
||
session_start_path.chmod(0o755)
|
||
|
||
update_path = hooks_dir / "crg-update.sh"
|
||
update_path.write_text(update_script, encoding="utf-8")
|
||
update_path.chmod(0o755)
|
||
|
||
hooks_obj = existing.get("hooks", {})
|
||
if not isinstance(hooks_obj, dict):
|
||
hooks_obj = {}
|
||
|
||
def _ensure_group(
|
||
event_name: str, matcher: str, hook_command: str, name: str, timeout: int,
|
||
) -> None:
|
||
arr = hooks_obj.get(event_name, [])
|
||
if not isinstance(arr, list):
|
||
arr = []
|
||
|
||
# De-duplicate by command (and type) inside nested hooks list.
|
||
def _group_has_command(group: Any) -> bool:
|
||
if not isinstance(group, dict):
|
||
return False
|
||
nested = group.get("hooks", [])
|
||
if not isinstance(nested, list):
|
||
return False
|
||
for h in nested:
|
||
if isinstance(h, dict) and h.get("type") == "command" \
|
||
and h.get("command") == hook_command:
|
||
return True
|
||
return False
|
||
|
||
if any(_group_has_command(g) for g in arr):
|
||
hooks_obj[event_name] = arr
|
||
return
|
||
|
||
arr.append(
|
||
{
|
||
"matcher": matcher,
|
||
"hooks": [
|
||
{
|
||
"type": "command",
|
||
"command": hook_command,
|
||
"name": name,
|
||
"timeout": timeout,
|
||
}
|
||
],
|
||
}
|
||
)
|
||
hooks_obj[event_name] = arr
|
||
|
||
_ensure_group(
|
||
event_name="SessionStart",
|
||
matcher="",
|
||
hook_command="bash .gemini/hooks/crg-session-start.sh",
|
||
name="code-review-graph status",
|
||
timeout=10_000,
|
||
)
|
||
_ensure_group(
|
||
event_name="AfterTool",
|
||
matcher="write_file|replace",
|
||
hook_command="bash .gemini/hooks/crg-update.sh",
|
||
name="code-review-graph update",
|
||
timeout=30_000,
|
||
)
|
||
|
||
existing["hooks"] = hooks_obj
|
||
settings_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
|
||
logger.info("Wrote Gemini CLI hooks config: %s", settings_path)
|
||
return settings_path
|
||
|
||
|
||
def install_gemini_cli_skills(repo_root: Path) -> Path:
|
||
"""Install Gemini CLI Agent Skills in .gemini/skills/<skill>/SKILL.md."""
|
||
skills_root = repo_root / ".gemini" / "skills"
|
||
skills_root.mkdir(parents=True, exist_ok=True)
|
||
|
||
for filename, skill in _SKILLS.items():
|
||
slug = filename.rsplit(".", 1)[0]
|
||
skill_dir = skills_root / slug
|
||
skill_dir.mkdir(parents=True, exist_ok=True)
|
||
skill_path = skill_dir / "SKILL.md"
|
||
content = (
|
||
"---\n"
|
||
f"name: {slug}\n"
|
||
f"description: {skill['description']}\n"
|
||
"---\n\n"
|
||
f"{skill['body']}\n"
|
||
)
|
||
skill_path.write_text(content, encoding="utf-8")
|
||
logger.info("Wrote Gemini CLI skill: %s", skill_path)
|
||
|
||
return skills_root
|
||
|
||
|
||
def inject_platform_instructions(repo_root: Path, target: str = "all") -> list[str]:
|
||
"""Inject 'use graph first' instructions into platform rule files.
|
||
|
||
Writes AGENTS.md, GEMINI.md, .cursorrules, and/or .windsurfrules
|
||
depending on ``target``:
|
||
|
||
- ``"all"`` (default): writes every file — matches pre-filter behavior.
|
||
- ``"claude"``: writes nothing (CLAUDE.md is handled by ``inject_claude_md``).
|
||
- any other platform key (``cursor``, ``windsurf``, ``antigravity``,
|
||
``opencode``): writes only the files associated with that platform.
|
||
|
||
Returns list of filenames that were created or updated.
|
||
"""
|
||
updated: list[str] = []
|
||
for filename, owners in _PLATFORM_INSTRUCTION_FILES.items():
|
||
if target != "all" and target not in owners:
|
||
continue
|
||
path = repo_root / filename
|
||
if filename in _PLATFORM_INSTRUCTION_CUSTOM_SECTIONS:
|
||
marker, section = _PLATFORM_INSTRUCTION_CUSTOM_SECTIONS[filename]
|
||
else:
|
||
marker, section = _CLAUDE_MD_SECTION_MARKER, _CLAUDE_MD_SECTION
|
||
if _inject_instructions(path, marker, section):
|
||
updated.append(filename)
|
||
return updated
|
||
|
||
|
||
# --- Cursor hooks ---
|
||
|
||
|
||
def generate_cursor_hooks_config() -> dict[str, Any]:
|
||
"""Generate Cursor hooks.json configuration.
|
||
|
||
Returns a dict conforming to the Cursor hooks schema (version 1) with
|
||
hooks for afterFileEdit, sessionStart, and beforeShellExecution.
|
||
Each hook points to a shell script in ~/.cursor/hooks/.
|
||
|
||
Returns:
|
||
Dict suitable for writing as ~/.cursor/hooks.json.
|
||
"""
|
||
hooks_dir = str(Path.home() / ".cursor" / "hooks")
|
||
return {
|
||
"version": 1,
|
||
"hooks": {
|
||
"afterFileEdit": [
|
||
{
|
||
"command": f"{hooks_dir}/crg-update.sh",
|
||
"timeout": 5,
|
||
},
|
||
],
|
||
"sessionStart": [
|
||
{
|
||
"command": f"{hooks_dir}/crg-session-start.sh",
|
||
"timeout": 5,
|
||
},
|
||
],
|
||
"beforeShellExecution": [
|
||
{
|
||
"matcher": "^git\\s+commit",
|
||
"command": f"{hooks_dir}/crg-pre-commit.sh",
|
||
"timeout": 10,
|
||
},
|
||
],
|
||
},
|
||
}
|
||
|
||
|
||
def _cursor_hook_scripts() -> dict[str, str]:
|
||
"""Return a mapping of filename -> shell script content for Cursor hooks.
|
||
|
||
Three scripts are generated:
|
||
- crg-update.sh: runs ``code-review-graph update --skip-flows`` after file edits
|
||
- crg-session-start.sh: runs ``code-review-graph status`` on session start
|
||
- crg-pre-commit.sh: runs ``code-review-graph detect-changes --brief`` before
|
||
git commit commands
|
||
|
||
All scripts:
|
||
- Read stdin (Cursor passes JSON context) and discard it
|
||
- Fail gracefully (exit 0) so they never block the editor
|
||
- Emit valid JSON on stdout per the Cursor hooks protocol
|
||
"""
|
||
update_script = """\
|
||
#!/usr/bin/env bash
|
||
# code-review-graph: auto-update graph after file edits (Cursor hook)
|
||
# Fails gracefully — never blocks the editor.
|
||
set -euo pipefail
|
||
|
||
# Consume stdin (Cursor sends JSON context)
|
||
cat > /dev/null
|
||
|
||
# Run update; swallow errors so the hook always succeeds.
|
||
output=$(code-review-graph update --skip-flows 2>&1) || true
|
||
|
||
# Emit valid JSON on stdout per Cursor hooks protocol.
|
||
python3 -c "
|
||
import json, sys
|
||
print(json.dumps({'message': 'graph updated', 'passed': True}))
|
||
" 2>/dev/null || echo '{"passed":true}'
|
||
|
||
exit 0
|
||
"""
|
||
|
||
session_start_script = """\
|
||
#!/usr/bin/env bash
|
||
# code-review-graph: show graph status on session start (Cursor hook)
|
||
# Fails gracefully — never blocks the editor.
|
||
set -euo pipefail
|
||
|
||
# Consume stdin
|
||
cat > /dev/null
|
||
|
||
# Capture status output
|
||
output=$(code-review-graph status 2>&1) || output="graph not built yet"
|
||
|
||
# Emit valid JSON on stdout
|
||
python3 -c "
|
||
import json, sys
|
||
msg = sys.stdin.read()
|
||
print(json.dumps({'message': msg, 'passed': True}))
|
||
" <<< "$output" 2>/dev/null || echo '{"passed":true}'
|
||
|
||
exit 0
|
||
"""
|
||
|
||
pre_commit_script = """\
|
||
#!/usr/bin/env bash
|
||
# code-review-graph: detect changes before git commit (Cursor hook)
|
||
# Fails gracefully — never blocks the editor.
|
||
set -euo pipefail
|
||
|
||
# Consume stdin
|
||
cat > /dev/null
|
||
|
||
# Run detect-changes; swallow errors
|
||
output=$(code-review-graph detect-changes --brief 2>&1) || output=""
|
||
|
||
# Emit valid JSON on stdout
|
||
python3 -c "
|
||
import json, sys
|
||
msg = sys.stdin.read()
|
||
print(json.dumps({'message': msg, 'passed': True}))
|
||
" <<< "$output" 2>/dev/null || echo '{"passed":true}'
|
||
|
||
exit 0
|
||
"""
|
||
|
||
return {
|
||
"crg-update.sh": update_script,
|
||
"crg-session-start.sh": session_start_script,
|
||
"crg-pre-commit.sh": pre_commit_script,
|
||
}
|
||
|
||
|
||
def install_cursor_hooks() -> Path:
|
||
"""Install Cursor hooks configuration and scripts at user level.
|
||
|
||
Writes ``~/.cursor/hooks.json`` (merging code-review-graph hooks
|
||
into any existing configuration) and creates executable shell scripts
|
||
in ``~/.cursor/hooks/``.
|
||
|
||
Returns:
|
||
Path to the hooks.json file that was written.
|
||
"""
|
||
cursor_dir = Path.home() / ".cursor"
|
||
hooks_json_path = cursor_dir / "hooks.json"
|
||
hooks_script_dir = cursor_dir / "hooks"
|
||
|
||
# --- Merge hooks.json ---
|
||
existing: dict[str, Any] = {}
|
||
if hooks_json_path.exists():
|
||
try:
|
||
existing = json.loads(hooks_json_path.read_text(encoding="utf-8"))
|
||
except (json.JSONDecodeError, OSError) as exc:
|
||
logger.warning("Could not read existing %s: %s", hooks_json_path, exc)
|
||
|
||
new_config = generate_cursor_hooks_config()
|
||
|
||
# Preserve version (use ours if absent)
|
||
existing.setdefault("version", new_config["version"])
|
||
|
||
# Merge hook arrays per event type
|
||
existing_hooks = existing.get("hooks", {})
|
||
if not isinstance(existing_hooks, dict):
|
||
existing_hooks = {}
|
||
|
||
for event, entries in new_config["hooks"].items():
|
||
event_hooks = existing_hooks.get(event, [])
|
||
if not isinstance(event_hooks, list):
|
||
event_hooks = []
|
||
# De-duplicate: skip if a hook with the same command already exists
|
||
existing_commands = {h.get("command", "") for h in event_hooks if isinstance(h, dict)}
|
||
for entry in entries:
|
||
if entry["command"] not in existing_commands:
|
||
event_hooks.append(entry)
|
||
existing_hooks[event] = event_hooks
|
||
|
||
existing["hooks"] = existing_hooks
|
||
|
||
cursor_dir.mkdir(parents=True, exist_ok=True)
|
||
hooks_json_path.write_text(
|
||
json.dumps(existing, indent=2) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
logger.info("Wrote Cursor hooks config: %s", hooks_json_path)
|
||
|
||
# --- Write hook scripts ---
|
||
hooks_script_dir.mkdir(parents=True, exist_ok=True)
|
||
scripts = _cursor_hook_scripts()
|
||
|
||
for filename, content in scripts.items():
|
||
script_path = hooks_script_dir / filename
|
||
script_path.write_text(content, encoding="utf-8")
|
||
# Make executable (owner rwx, group rx, other rx)
|
||
script_path.chmod(stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH)
|
||
logger.info("Wrote Cursor hook script: %s", script_path)
|
||
|
||
return hooks_json_path
|
||
|
||
|
||
def install_qoder_skills(repo_root: Path) -> Path | None:
|
||
"""Install skills to Qoder's project-level skills directory.
|
||
|
||
Qoder expects skills in .qoder/skills/{skillName}/SKILL.md format within the project.
|
||
This function copies the project's skills/ directory contents to that location.
|
||
|
||
Args:
|
||
repo_root: Repository root directory (where the skills/ folder is located).
|
||
|
||
Returns:
|
||
Path to the Qoder skills directory, or None if installation failed.
|
||
"""
|
||
# Qoder skills directory (project-level)
|
||
qoder_skills_dir = repo_root / ".qoder" / "skills"
|
||
qoder_skills_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Source skills directory in the project
|
||
source_skills_dir = repo_root / "skills"
|
||
if not source_skills_dir.exists():
|
||
logger.warning("No skills/ directory found in %s", repo_root)
|
||
return None
|
||
|
||
installed_count = 0
|
||
for skill_dir in source_skills_dir.iterdir():
|
||
if skill_dir.is_dir():
|
||
skill_file = skill_dir / "SKILL.md"
|
||
if skill_file.exists():
|
||
target_dir = qoder_skills_dir / skill_dir.name
|
||
target_dir.mkdir(parents=True, exist_ok=True)
|
||
target_file = target_dir / "SKILL.md"
|
||
target_file.write_text(skill_file.read_text(encoding="utf-8"), encoding="utf-8")
|
||
logger.info("Installed Qoder skill: %s", skill_dir.name)
|
||
installed_count += 1
|
||
|
||
if installed_count > 0:
|
||
logger.info("Installed %d skill(s) to %s", installed_count, qoder_skills_dir)
|
||
return qoder_skills_dir
|
||
return None
|
||
|
||
|
||
# --- OpenCode plugin ---
|
||
|
||
|
||
def _opencode_plugin_content() -> str:
|
||
"""Return TypeScript source for the OpenCode user-level plugin.
|
||
|
||
The plugin hooks into three OpenCode events to mirror the Claude Code
|
||
hook behaviors:
|
||
|
||
1. ``file.edited`` — runs ``code-review-graph update --skip-flows``
|
||
2. ``session.created`` — runs ``code-review-graph status``
|
||
3. ``tool.execute.before`` — when the tool is a shell command starting
|
||
with ``git commit``, runs ``code-review-graph detect-changes --brief``
|
||
|
||
All handlers use try/catch so errors never break the editor session.
|
||
The plugin uses Bun's ``$`` shell API (provided by OpenCode's plugin
|
||
context) for subprocess execution.
|
||
"""
|
||
return """\
|
||
import type { Plugin } from "@opencode-ai/plugin"
|
||
|
||
/**
|
||
* code-review-graph plugin for OpenCode.
|
||
*
|
||
* Keeps the knowledge graph up-to-date and surfaces status
|
||
* information automatically during coding sessions.
|
||
*
|
||
* Installed by: code-review-graph install --platform opencode
|
||
*/
|
||
|
||
// Helper: run a shell command quietly, swallowing errors.
|
||
async function run($: any, cmd: string): Promise<string> {
|
||
try {
|
||
const result = await $`${cmd}`.quiet()
|
||
return result.stdout?.toString().trim() ?? ""
|
||
} catch {
|
||
return ""
|
||
}
|
||
}
|
||
|
||
export default (app: any) => {
|
||
// 1. Auto-update graph after file edits
|
||
app.on("file.edited", async ({ $ }: { $: any }) => {
|
||
try {
|
||
await $`code-review-graph update --skip-flows`.quiet()
|
||
} catch {
|
||
// Swallow — graph may not be built yet for this project.
|
||
}
|
||
})
|
||
|
||
// 2. Show graph status when a new session starts
|
||
app.on("session.created", async ({ $ }: { $: any }) => {
|
||
try {
|
||
const result = await $`code-review-graph status`.quiet()
|
||
const output = result.stdout?.toString().trim()
|
||
if (output) {
|
||
console.log("[code-review-graph]", output)
|
||
}
|
||
} catch {
|
||
// Swallow — not every project has a graph.
|
||
}
|
||
})
|
||
|
||
// 3. Detect changes before git commit commands
|
||
app.on("tool.execute.before", async (ctx: any) => {
|
||
try {
|
||
const input = ctx?.input ?? ctx?.params ?? {}
|
||
const cmd =
|
||
input.command ?? input.cmd ?? input.content ?? ""
|
||
if (typeof cmd === "string" && /^git\\s+commit/i.test(cmd)) {
|
||
const result =
|
||
await ctx.$`code-review-graph detect-changes --brief`.quiet()
|
||
const output = result.stdout?.toString().trim()
|
||
if (output) {
|
||
console.log("[code-review-graph] Pre-commit analysis:\\n" + output)
|
||
}
|
||
}
|
||
} catch {
|
||
// Swallow — never block a commit.
|
||
}
|
||
})
|
||
}
|
||
"""
|
||
|
||
|
||
def install_opencode_plugin() -> Path:
|
||
"""Install the OpenCode user-level plugin for code-review-graph.
|
||
|
||
Writes ``~/.config/opencode/plugins/crg-plugin.ts``. Creates the
|
||
directories if they don't exist. If the file already exists it is
|
||
overwritten (the plugin is self-contained and idempotent).
|
||
|
||
Returns:
|
||
Path to the plugin file that was written.
|
||
"""
|
||
plugins_dir = Path.home() / ".config" / "opencode" / "plugins"
|
||
plugin_path = plugins_dir / "crg-plugin.ts"
|
||
|
||
plugins_dir.mkdir(parents=True, exist_ok=True)
|
||
plugin_path.write_text(_opencode_plugin_content(), encoding="utf-8")
|
||
logger.info("Wrote OpenCode plugin: %s", plugin_path)
|
||
|
||
return plugin_path
|