Merge pull request #372 from AIOSAI/citizen/seedgo-hook-consolidation-track-a

feat(hooks): DPLAN-0139 Track A — hook consolidation
This commit is contained in:
AIPass
2026-04-21 08:28:59 -07:00
committed by GitHub
11 changed files with 389 additions and 656 deletions
@@ -1,310 +0,0 @@
#!/usr/bin/env python3
"""
Silent Auto-fix Hook - Runs ACTUAL validation and tells Claude to fix silently.
Key behaviors:
- Runs real linters (ruff, py_compile) not just pattern lists
- Validates JSON with json.load()
- Outputs via additionalContext so Claude sees errors
- Claude fixes silently without announcing
- Simple indicator for user console
- Smart batching per-file
Version: 4.3.0
"""
import json
import sys
import subprocess
from pathlib import Path
EDIT_TOOLS = ["Edit", "Write", "MultiEdit", "NotebookEdit"]
LAST_FILE_PATH = Path(__file__).parent / ".last_diagnostics_file"
SKIP_EXTENSIONS = {".md", ".txt", ".log", ".csv", ".html"}
# AIPass-specific Python patterns to check
PYTHON_PATTERNS = {
"bad_optional": {
"pattern": ": str = None",
"message": "Optional param should use 'str | None = None' pattern"
},
"logger_debug": {
"pattern": "logger.debug(",
"message": "Use logger.info for SystemLogger (logger.debug not supported)"
},
"return_error_msg": {
"pattern": "return error_msg",
"message": "Return None for error states, not error_msg string"
},
"open_no_encoding": {
"pattern": "open(",
"requires_missing": "encoding=",
"message": "open() without encoding='utf-8'"
},
"log_not_log_operation": {
"pattern": ".log(",
"message": "Use log_operation() with success/error params, not .log()"
},
"dict_none_no_check": {
"pattern": "Dict | None",
"message": "Dict | None return: Add None check before using (if result is None: return)"
}
}
# JSON-specific patterns for emoji corruption
JSON_CORRUPTION_CHARS = ['\ufffd', '\x00']
def run_python_checks(file_path: str) -> list[str]:
"""Run actual Python validation - returns list of errors."""
errors = []
# 1. Syntax check with py_compile
try:
result = subprocess.run(
[sys.executable, "-m", "py_compile", file_path],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
errors.append(f"SYNTAX: {result.stderr.strip()}")
except Exception:
pass
# 2. Ruff check (if available) - fast linter
try:
result = subprocess.run(
["ruff", "check", "--select=E,F,W", "--output-format=text", file_path],
capture_output=True,
text=True,
timeout=10
)
if result.stdout.strip():
for line in result.stdout.strip().split("\n")[:5]: # Max 5 errors
errors.append(f"LINT: {line}")
except FileNotFoundError:
pass # ruff not installed
except Exception:
pass
# 3. AIPass-specific pattern checks
try:
content = Path(file_path).read_text(encoding="utf-8")
lines = content.split("\n")
for check in PYTHON_PATTERNS.values():
pattern = check["pattern"]
message = check["message"]
requires_missing = check.get("requires_missing")
# For patterns that require something to be missing
if requires_missing:
if pattern in content and requires_missing not in content:
errors.append(f"PATTERN: {message}")
continue
# Standard pattern check - scan lines
for line in lines:
stripped = line.strip()
# Skip comments and strings
if stripped.startswith(("#", '"', "'")):
continue
# Skip if pattern appears in a string on this line
if f'"{pattern}' in line or f"'{pattern}" in line:
continue
if pattern in line:
errors.append(f"PATTERN: {message}")
break # One error per pattern type
except Exception:
pass
return errors
def run_json_checks(file_path: str) -> list[str]:
"""Run actual JSON validation - returns list of errors."""
errors = []
try:
content = Path(file_path).read_text(encoding="utf-8")
# Check for emoji corruption before parsing
for char in JSON_CORRUPTION_CHARS:
if char in content:
errors.append(f"EMOJI CORRUPTION: Found corrupted character '{repr(char)}' - check allowed_emojis arrays")
break
# Try to parse JSON
try:
data = json.loads(content)
# Check for corruption in emoji arrays specifically
if isinstance(data, dict):
for key in ['allowed_emojis', 'emojis', 'emoji_list']:
if key in data and isinstance(data[key], list):
for item in data[key]:
if isinstance(item, str) and len(item) == 1:
if ord(item) < 128 and item not in '\u2713\u2717':
errors.append(f"EMOJI CORRUPTION: Suspicious char '{item}' in {key} - may be corrupted emoji")
break
except json.JSONDecodeError as e:
errors.append(f"JSON SYNTAX: {e.msg} at line {e.lineno}")
except Exception as e:
errors.append(f"READ ERROR: {e!s}")
return errors
def run_seedgo_checklist(file_path: str) -> list[str]:
"""Run seedgo standards checklist — returns violations only."""
# Skip Claude hooks - they don't follow project standards
if '/.claude/hooks/' in file_path:
return []
try:
result = subprocess.run(
["drone", "@seedgo", "checklist", file_path],
capture_output=True,
text=True,
timeout=15,
cwd=str(Path.home() / "Projects" / "AIPass")
)
if result.returncode != 0:
return [] # Command failed, skip
violations = []
current_standard = None
for line in result.stdout.split("\n"):
line = line.strip()
# Checklist output format: "✗ standard_name: detail"
if line.startswith("\u2717"):
violation = line[1:].strip()
if violation:
violations.append(violation)
return violations[:5] # Top 5 only
except FileNotFoundError:
return [] # drone not available
except Exception:
return []
def should_skip_file(file_path: str) -> bool:
"""Check if file should be skipped."""
if not file_path:
return True
ext = Path(file_path).suffix.lower()
return ext in SKIP_EXTENSIONS
def is_same_file_as_last(file_path: str) -> bool:
"""Smart batching - skip if same file as last check."""
try:
if LAST_FILE_PATH.exists():
last = LAST_FILE_PATH.read_text(encoding="utf-8").strip()
if last == file_path:
return True
LAST_FILE_PATH.write_text(file_path, encoding="utf-8")
except Exception:
pass
return False
def main():
"""Main hook entry point."""
try:
input_data = json.load(sys.stdin)
tool_name = input_data.get("tool_name", "")
tool_input = input_data.get("tool_input", {})
file_path = tool_input.get("file_path", "")
# Only process edit tools
if tool_name not in EDIT_TOOLS:
return
# Skip non-code files
if should_skip_file(file_path):
return
# Smart batching - don't re-check same file
if is_same_file_as_last(file_path):
return
# Determine file type and run checks
errors = []
file_type = ""
if file_path.endswith(".py"):
file_type = "Python"
errors = run_python_checks(file_path)
# Seedgo standards checklist (Python files only)
seedgo_violations = run_seedgo_checklist(file_path)
for v in seedgo_violations:
errors.append(f"SEEDGO: {v}")
elif file_path.endswith(".json"):
file_type = "JSON"
errors = run_json_checks(file_path)
else:
return # No validation for other types yet
# Build context message
context = ""
ide_instruction = f"\n\nAlso run mcp__ide__getDiagnostics on {file_path} to check for type errors. Fix any errors found silently."
if errors:
error_text = "\n".join(f" - {e}" for e in errors)
context = f"""[AUTO-FIX REQUIRED] {len(errors)} validation error(s) in {Path(file_path).name}:
{error_text}
MANDATORY: You MUST fix every error listed above in {Path(file_path).name} before proceeding with any other work. Edit the file now to resolve these issues. Do not skip, defer, or ignore — fix them immediately, then continue with your task."""
# For Python files, also check IDE diagnostics for type errors
if file_path.endswith(".py"):
if errors:
context += ide_instruction
else:
# No other errors, but still check IDE diagnostics
context = f"[TYPE-CHECK] Run mcp__ide__getDiagnostics on {file_path} and fix any type errors silently."
errors = ["IDE diagnostics check requested"]
output = {
"hookSpecificOutput": {
"hookEventName": "PostToolUse",
"additionalContext": context
},
"systemMessage": f"[AUTO-FIX] {len(errors)} issue(s) — fix before continuing"
}
print(json.dumps(output))
# Exit 0 so JSON additionalContext is delivered to the model
elif errors:
# JSON or other file with errors
output = {
"hookSpecificOutput": {
"hookEventName": "PostToolUse",
"additionalContext": context
},
"systemMessage": f"[AUTO-FIX] {len(errors)} issue(s) — fix before continuing"
}
print(json.dumps(output))
# Exit 0 so JSON additionalContext is delivered to the model
else:
# No errors - just tiny indicator
output = {
"systemMessage": "[diagnostics] ok"
}
print(json.dumps(output))
except Exception:
pass # Silent fail
if __name__ == "__main__":
main()
-5
View File
@@ -1,5 +0,0 @@
#!/bin/bash
# Log hook activity for statusline display
# Usage: source from hook commands or call directly
# hook_logger.sh <hook_name>
echo "$(date +%s) $1" > /tmp/aipass-hook-last
@@ -1,37 +0,0 @@
#!/usr/bin/env python3
"""Notification Hook — Plays sound when AI needs permission."""
import json
import sys
import subprocess
from pathlib import Path
SOUNDS_DIR = Path(__file__).parent.parent / "sounds"
SOUND_FILE = SOUNDS_DIR / "mixkit-clear-announce-tones-2861.wav"
def play_sound() -> None:
if not SOUND_FILE.exists():
return
try:
subprocess.Popen(
["aplay", "-q", str(SOUND_FILE)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception:
pass
def main():
try:
hook_data = json.loads(sys.stdin.read())
if hook_data.get("hook_event_name") == "Notification":
play_sound()
except Exception:
pass
sys.exit(0)
if __name__ == "__main__":
main()
-125
View File
@@ -1,125 +0,0 @@
#!/bin/bash
# AIPass statusline — pretty, context-aware
input=$(cat)
# Extract fields
dir=$(echo "$input" | jq -r '.workspace.current_dir // ""')
model=$(echo "$input" | jq -r '.model.display_name // "?"')
ctx=$(echo "$input" | jq -r '.context_window.remaining_percentage // empty')
cost=$(echo "$input" | jq -r '.cost.total_cost_usd // empty')
lines_add=$(echo "$input" | jq -r '.cost.total_lines_added // 0')
lines_rm=$(echo "$input" | jq -r '.cost.total_lines_removed // 0')
session=$(echo "$input" | jq -r '.session_name // empty')
# Colors
RST='\033[0m'
DIM='\033[2m'
BOLD='\033[1m'
GREEN='\033[32m'
YELLOW='\033[33m'
RED='\033[31m'
CYAN='\033[36m'
MAGENTA='\033[35m'
WHITE='\033[97m'
# Shorten directory — extract branch name if inside AIPass
branch=""
if [[ "$dir" == *"/src/aipass/"* ]]; then
branch=$(echo "$dir" | sed 's|.*/src/aipass/||' | cut -d/ -f1)
elif [[ "$dir" == *"/src/commons"* ]]; then
branch="commons"
elif [[ "$dir" == *"/src/skills"* ]]; then
branch="skills"
elif [[ "$dir" == *"/AIPass"* ]]; then
branch="root"
fi
# Context color — green > 50%, yellow 20-50%, red < 20%
ctx_color="$GREEN"
if [ -n "$ctx" ]; then
if [ "$ctx" -lt 20 ] 2>/dev/null; then
ctx_color="$RED"
elif [ "$ctx" -lt 50 ] 2>/dev/null; then
ctx_color="$YELLOW"
fi
fi
# Build context bar (10 chars wide)
bar=""
if [ -n "$ctx" ]; then
used=$((100 - ctx))
filled=$((used / 10))
empty=$((10 - filled))
bar="${DIM}["
for ((i=0; i<filled; i++)); do bar+="█"; done
for ((i=0; i<empty; i++)); do bar+="░"; done
bar+="]${RST}"
fi
# Shorten model name
short_model=$(echo "$model" | sed 's/Claude //')
# Git branch
git_branch=$(git -C "$dir" branch --show-current 2>/dev/null)
# Build output
out=""
# Branch or directory
if [ -n "$branch" ]; then
out+="${CYAN}@${branch}${RST}"
else
short_dir=$(echo "$dir" | sed "s|$HOME|~|" | awk -F/ '{if(NF>=2) print $(NF-1)"/"$NF; else print $NF}')
out+="${DIM}${short_dir}${RST}"
fi
# Git branch
if [ -n "$git_branch" ]; then
out+=" ${DIM}(${RST}${YELLOW}${git_branch}${RST}${DIM})${RST}"
fi
# Separator
out+=" ${DIM}│${RST} "
# Model
out+="${MAGENTA}${short_model}${RST}"
# Separator
out+=" ${DIM}│${RST} "
# Context
if [ -n "$ctx" ]; then
out+="${ctx_color}${ctx}%${RST} ${bar}"
else
out+="${DIM}...${RST}"
fi
# Cost (if any)
if [ -n "$cost" ] && [ "$cost" != "0" ]; then
cost_fmt=$(printf "%.2f" "$cost" 2>/dev/null || echo "$cost")
out+=" ${DIM}│${RST} ${DIM}\$${cost_fmt}${RST}"
fi
# Lines changed
if [ "$lines_add" -gt 0 ] 2>/dev/null || [ "$lines_rm" -gt 0 ] 2>/dev/null; then
out+=" ${DIM}│${RST} ${GREEN}+${lines_add}${RST}${DIM}/${RST}${RED}-${lines_rm}${RST}"
fi
# Session name if set
if [ -n "$session" ]; then
out=" ${DIM}[${RST}${WHITE}${session}${RST}${DIM}]${RST} ${out}"
fi
# Hook activity — show last hook if fired within 3 seconds
HOOK_FILE="/tmp/aipass-hook-last"
if [ -f "$HOOK_FILE" ]; then
hook_data=$(cat "$HOOK_FILE" 2>/dev/null)
hook_ts=$(echo "$hook_data" | cut -d' ' -f1)
hook_name=$(echo "$hook_data" | cut -d' ' -f2-)
now=$(date +%s)
if [ -n "$hook_ts" ] && [ $((now - hook_ts)) -le 3 ] 2>/dev/null; then
out+=" ${DIM}│${RST} ${DIM}hook:${RST}${YELLOW}${hook_name}${RST}"
fi
fi
printf '%b' "$out"
-38
View File
@@ -1,38 +0,0 @@
#!/usr/bin/env python3
"""Stop Hook — Plays achievement bell when AI finishes responding."""
import json
import sys
import subprocess
from pathlib import Path
SOUNDS_DIR = Path(__file__).parent.parent / "sounds"
SOUND_FILE = SOUNDS_DIR / "mixkit-achievement-bell-600.wav"
def play_sound() -> None:
if not SOUND_FILE.exists():
return
try:
subprocess.Popen(
["aplay", "-q", str(SOUND_FILE)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception:
pass
def main():
try:
hook_data = json.loads(sys.stdin.read())
if hook_data.get("hook_event_name") == "Stop":
if not hook_data.get("stop_hook_active", False):
play_sound()
except Exception:
pass
sys.exit(0)
if __name__ == "__main__":
main()
-100
View File
@@ -1,100 +0,0 @@
#!/usr/bin/env python3
"""
SubagentStop Gate — Checks files modified by subagents before allowing them to finish.
Runs seedgo checklist + basic validation on any .py files the subagent touched.
If violations found, blocks the stop and tells the subagent to fix them.
Version: 1.0.0
"""
import json
import sys
import subprocess
from pathlib import Path
AIPASS_ROOT = Path.home() / "Projects" / "AIPass"
def get_modified_py_files() -> list[str]:
"""Get Python files modified in the working tree (unstaged + staged)."""
try:
result = subprocess.run(
["git", "diff", "--name-only", "HEAD"],
capture_output=True, text=True, timeout=5,
cwd=str(AIPASS_ROOT)
)
files = []
for line in result.stdout.strip().split("\n"):
line = line.strip()
if line.endswith(".py") and not line.startswith(".claude/"):
full = AIPASS_ROOT / line
if full.exists():
files.append(str(full))
return files
except Exception:
return []
def run_seedgo_checklist(file_path: str) -> list[str]:
"""Run seedgo checklist on a single file."""
if "/.claude/" in file_path:
return []
try:
result = subprocess.run(
["drone", "@seedgo", "checklist", file_path],
capture_output=True, text=True, timeout=15,
cwd=str(AIPASS_ROOT)
)
if result.returncode != 0:
return []
violations = []
for line in result.stdout.split("\n"):
line = line.strip()
if line.startswith("\u2717"):
v = line[1:].strip()
if v:
violations.append(v)
return violations[:5]
except Exception:
return []
def main():
try:
input_data = json.load(sys.stdin)
modified = get_modified_py_files()
if not modified:
return # Nothing to check
all_violations = {}
for f in modified:
vs = run_seedgo_checklist(f)
if vs:
name = Path(f).name
all_violations[name] = vs
if not all_violations:
return # All clear
# Build the block reason
lines = ["Standards violations found in files you modified:\n"]
for fname, vs in all_violations.items():
lines.append(f" {fname}:")
for v in vs:
lines.append(f" - {v}")
lines.append("\nFix these violations before finishing.")
output = {
"decision": "block",
"reason": "\n".join(lines)
}
print(json.dumps(output))
except Exception:
pass # Silent fail — don't block on errors
if __name__ == "__main__":
main()
-40
View File
@@ -1,40 +0,0 @@
#!/usr/bin/env python3
"""Tool Use Hook — Plays key press sound when AI uses tools."""
import json
import sys
import subprocess
from pathlib import Path
SOUNDS_DIR = Path(__file__).parent.parent / "sounds"
SOUND_FILE = SOUNDS_DIR / "mixkit-atm-cash-machine-key-press-2841.wav"
SOUND_TOOLS = ["Bash", "Edit", "MultiEdit", "Write", "Read", "Grep", "Glob"]
def play_sound() -> None:
if not SOUND_FILE.exists():
return
try:
subprocess.Popen(
["aplay", "-q", str(SOUND_FILE)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception:
pass
def main():
try:
hook_data = json.loads(sys.stdin.read())
if hook_data.get("hook_event_name") == "PreToolUse":
if hook_data.get("tool_name", "") in SOUND_TOOLS:
play_sound()
except Exception:
pass
sys.exit(0)
if __name__ == "__main__":
main()
Submodule .claude/worktrees/agent-aa2b7d43 deleted from 98191fa97e
+2
View File
@@ -0,0 +1,2 @@
# Include hooks directory for pip packaging
recursive-include .claude/hooks *.py *.md
+3
View File
@@ -64,6 +64,9 @@ aipass = "aipass.cli:cli_entry"
[tool.hatch.build.targets.wheel]
packages = ["src/aipass"]
[tool.hatch.build.targets.wheel.force-include]
".claude/hooks" = "aipass/_hooks"
[tool.pytest.ini_options]
testpaths = ["tests", "src"]
norecursedirs = ["templates", "*.egg-info", ".git", ".venv", "__pycache__"]
@@ -0,0 +1,384 @@
"""Tests for Track A hook scripts — auto_fix_diagnostics, pre_edit_gate, subagent_stop_gate.
# =================== META ====================
# Name: test_hooks_track_a.py
# Description: Track A hook tests — DPLAN-0139 coverage for auto_fix, pre_edit_gate, subagent_stop
# Version: 1.0.0
# Created: 2026-04-20
# Modified: 2026-04-20
# =============================================
"""
import importlib.util
import io
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# ---------------------------------------------------------------------------
# Helpers — hook loader
# ---------------------------------------------------------------------------
def _find_repo_root() -> Path:
"""Walk up from this file to find the git repo root."""
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
return parent
return Path(__file__).resolve().parents[4] # fallback
HOOKS_DIR = _find_repo_root() / ".claude" / "hooks"
def _load_hook(name: str):
"""Import a hook script by filename via importlib (outside package)."""
path = HOOKS_DIR / name
if not path.exists():
pytest.skip(f"Hook script not found: {path}")
spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
return mod
# ---------------------------------------------------------------------------
# auto_fix_diagnostics.py tests
# ---------------------------------------------------------------------------
def test_auto_fix_skips_non_edit_tool(capsys):
"""stdin with tool_name=Read → no output (not an edit tool)."""
mod = _load_hook("auto_fix_diagnostics.py")
payload = json.dumps({"tool_name": "Read", "tool_input": {"file_path": "/tmp/foo.py"}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_auto_fix_skips_non_py_file(capsys):
"""stdin with tool_name=Edit, file_path ending .md → no output (skipped extension)."""
mod = _load_hook("auto_fix_diagnostics.py")
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/foo.md"}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_auto_fix_does_not_crash_empty_stdin():
"""Empty/malformed JSON on stdin → no exception raised."""
mod = _load_hook("auto_fix_diagnostics.py")
with patch("sys.stdin", io.StringIO("")):
mod.main() # must not raise
def test_auto_fix_does_not_crash_missing_fields():
"""Valid JSON but no tool_name → no exception raised."""
mod = _load_hook("auto_fix_diagnostics.py")
payload = json.dumps({"some_other_key": "value"})
with patch("sys.stdin", io.StringIO(payload)):
mod.main() # must not raise
def test_auto_fix_silent_fix_label(tmp_path, capsys):
"""When run_python_checks returns an error, output contains [SILENT-FIX] not [AUTO-FIX]."""
mod = _load_hook("auto_fix_diagnostics.py")
# Point to a real .py file so extension check passes
fake_py = tmp_path / "fake.py"
fake_py.write_text("x = 1\n", encoding="utf-8")
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": str(fake_py)}})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "run_python_checks", return_value=["LINT: E501 line too long"]),
patch.object(mod, "run_seedgo_checklist", return_value=[]),
patch.object(mod, "run_pyright_check", return_value=([], False)),
patch.object(mod, "run_ruff_lint_structured", return_value=[]),
patch.object(mod, "save_diagnostics_state"),
):
mod.main()
captured = capsys.readouterr()
assert captured.out.strip() != ""
output = json.loads(captured.out)
context = output["hookSpecificOutput"]["additionalContext"]
assert "[SILENT-FIX]" in context
assert "[AUTO-FIX]" not in context
assert "[SILENT-FIX]" in output["systemMessage"]
def test_auto_fix_fallback_message_in_context(tmp_path, capsys):
"""When pyright is unavailable (fallback_needed=True), context contains [FALLBACK] text."""
mod = _load_hook("auto_fix_diagnostics.py")
fake_py = tmp_path / "check.py"
fake_py.write_text("x = 1\n", encoding="utf-8")
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": str(fake_py)}})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "run_python_checks", return_value=["LINT: E501 line too long"]),
patch.object(mod, "run_seedgo_checklist", return_value=[]),
patch.object(mod, "run_pyright_check", return_value=([], True)),
patch.object(mod, "run_ruff_lint_structured", return_value=[]),
patch.object(mod, "save_diagnostics_state"),
):
mod.main()
captured = capsys.readouterr()
output = json.loads(captured.out)
context = output["hookSpecificOutput"]["additionalContext"]
assert "[FALLBACK]" in context
assert "mcp__ide__getDiagnostics" in context
def test_auto_fix_fallback_stderr_announce(tmp_path, capsys):
"""When fallback_needed=True, [FALLBACK ENGAGED] is printed to stderr."""
mod = _load_hook("auto_fix_diagnostics.py")
fake_py = tmp_path / "check2.py"
fake_py.write_text("x = 1\n", encoding="utf-8")
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": str(fake_py)}})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "run_python_checks", return_value=[]),
patch.object(mod, "run_seedgo_checklist", return_value=[]),
patch.object(mod, "run_pyright_check", return_value=([], True)),
patch.object(mod, "run_ruff_lint_structured", return_value=[]),
patch.object(mod, "save_diagnostics_state"),
):
mod.main()
captured = capsys.readouterr()
assert "[FALLBACK ENGAGED" in captured.err
def test_run_pyright_check_returns_tuple_on_file_not_found():
"""run_pyright_check returns ([], True) when pyright binary is missing."""
mod = _load_hook("auto_fix_diagnostics.py")
with patch("subprocess.run", side_effect=FileNotFoundError("pyright not found")):
result = mod.run_pyright_check("/tmp/some_file.py")
assert result == ([], True)
def test_run_pyright_check_returns_false_on_timeout():
"""run_pyright_check returns ([], False) on timeout — no fallback signal."""
import subprocess
mod = _load_hook("auto_fix_diagnostics.py")
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("pyright", 15)):
result = mod.run_pyright_check("/tmp/some_file.py")
assert result == ([], False)
def test_run_pyright_check_skips_hook_files():
"""run_pyright_check returns ([], False) for hook files without calling subprocess."""
mod = _load_hook("auto_fix_diagnostics.py")
mock_run = MagicMock()
with patch("subprocess.run", mock_run):
result = mod.run_pyright_check("/home/user/.claude/hooks/some_hook.py")
mock_run.assert_not_called()
assert result == ([], False)
# ---------------------------------------------------------------------------
# pre_edit_gate.py tests
# (pre_edit_gate lives in ~/.claude/hooks — load from global location)
# ---------------------------------------------------------------------------
GLOBAL_HOOKS_DIR = Path.home() / ".claude" / "hooks"
def _load_global_hook(name: str):
"""Import a hook script from the global ~/.claude/hooks/ directory."""
path = GLOBAL_HOOKS_DIR / name
if not path.exists():
pytest.skip(f"Global hook script not found: {path}")
spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
return mod
def test_gate_allows_no_state_file(tmp_path, capsys, monkeypatch):
"""No state file present → no output (allow)."""
mod = _load_global_hook("pre_edit_gate.py")
monkeypatch.setattr(mod, "STATE_FILE", tmp_path / "no_such_state.json")
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/seedgo/foo.py"}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_allows_same_file(tmp_path, capsys, monkeypatch):
"""State has error for file A, editing file A → allow (no block output)."""
mod = _load_global_hook("pre_edit_gate.py")
errored_file = tmp_path / "foo.py"
errored_file.write_text("x = 1\n", encoding="utf-8")
state = {"file": str(errored_file), "errors": [{"line": 1, "message": "some error"}]}
state_file = tmp_path / "state.json"
state_file.write_text(json.dumps(state), encoding="utf-8")
monkeypatch.setattr(mod, "STATE_FILE", state_file)
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": str(errored_file)}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_blocks_different_file_same_branch(tmp_path, capsys, monkeypatch):
"""State has error for seedgo/foo.py, editing seedgo/bar.py → block + sys.exit(2)."""
mod = _load_global_hook("pre_edit_gate.py")
# Use real AIPass src paths so _get_branch works
aipass_src = Path("/home/patrick/Projects/AIPass/src/aipass")
errored_file = str(aipass_src / "seedgo" / "foo.py")
edit_target = str(aipass_src / "seedgo" / "bar.py")
state = {"file": errored_file, "errors": [{"line": 5, "message": "type error here"}]}
state_file = tmp_path / "state.json"
state_file.write_text(json.dumps(state), encoding="utf-8")
monkeypatch.setattr(mod, "STATE_FILE", state_file)
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": edit_target}})
with patch("sys.stdin", io.StringIO(payload)):
with pytest.raises(SystemExit) as exc_info:
mod.main()
assert exc_info.value.code == 2
captured = capsys.readouterr()
assert captured.out.strip() != ""
output = json.loads(captured.out)
assert output["decision"] == "block"
def test_gate_allows_different_branch(tmp_path, capsys, monkeypatch):
"""State has error for flow/foo.py, editing seedgo/bar.py → allow (different branches)."""
mod = _load_global_hook("pre_edit_gate.py")
aipass_src = Path("/home/patrick/Projects/AIPass/src/aipass")
errored_file = str(aipass_src / "flow" / "foo.py")
edit_target = str(aipass_src / "seedgo" / "bar.py")
state = {"file": errored_file, "errors": [{"line": 1, "message": "error in flow"}]}
state_file = tmp_path / "state.json"
state_file.write_text(json.dumps(state), encoding="utf-8")
monkeypatch.setattr(mod, "STATE_FILE", state_file)
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": edit_target}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_does_not_crash_malformed_stdin(monkeypatch, tmp_path):
"""Bad JSON on stdin → no exception raised."""
mod = _load_global_hook("pre_edit_gate.py")
monkeypatch.setattr(mod, "STATE_FILE", tmp_path / "no_state.json")
with patch("sys.stdin", io.StringIO("not json!!!!")):
mod.main() # must not raise
def test_gate_allows_non_edit_tool(tmp_path, capsys, monkeypatch):
"""Non-edit tool_name → no output regardless of state."""
mod = _load_global_hook("pre_edit_gate.py")
aipass_src = Path("/home/patrick/Projects/AIPass/src/aipass")
errored_file = str(aipass_src / "seedgo" / "foo.py")
edit_target = str(aipass_src / "seedgo" / "bar.py")
state = {"file": errored_file, "errors": [{"line": 1, "message": "error"}]}
state_file = tmp_path / "state.json"
state_file.write_text(json.dumps(state), encoding="utf-8")
monkeypatch.setattr(mod, "STATE_FILE", state_file)
payload = json.dumps({"tool_name": "Read", "tool_input": {"file_path": edit_target}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
# ---------------------------------------------------------------------------
# subagent_stop_gate.py tests
# ---------------------------------------------------------------------------
def test_subagent_gate_no_crash_empty_stdin():
"""Bad JSON on stdin → no exception raised."""
mod = _load_hook("subagent_stop_gate.py")
with patch("sys.stdin", io.StringIO("bad json")):
mod.main() # must not raise
def test_subagent_gate_no_crash_no_files(capsys):
"""Valid stdin but no modified files → no exception, no block output."""
mod = _load_hook("subagent_stop_gate.py")
payload = json.dumps({"stop_hook_active": True})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "get_modified_py_files", return_value=[]),
):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_subagent_gate_no_block_when_no_violations(capsys):
"""Modified files present but no seedgo violations → no block output."""
mod = _load_hook("subagent_stop_gate.py")
payload = json.dumps({"stop_hook_active": True})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "get_modified_py_files", return_value=["/tmp/foo.py"]),
patch.object(mod, "run_seedgo_checklist", return_value=[]),
):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_subagent_gate_blocks_on_violations(capsys):
"""Modified files with seedgo violations → block decision in output."""
mod = _load_hook("subagent_stop_gate.py")
payload = json.dumps({"stop_hook_active": True})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "get_modified_py_files", return_value=["/tmp/foo.py"]),
patch.object(mod, "run_seedgo_checklist", return_value=["open() without encoding='utf-8'"]),
):
mod.main()
captured = capsys.readouterr()
assert captured.out.strip() != ""
output = json.loads(captured.out)
assert output["decision"] == "block"