scripts/pre_commit_gate.py - master - davydov/NGPowers

scripts/pre_commit_gate.py
208 строк10 KB
Roman Davydov
feat(ngpowers): implement pre-archival safety gate and task auto-archiving safety
9 часов назад
9 часов назадadd1f7e
#!/usr/bin/env python3
import os
import sys
import sqlite3
import subprocess

def get_staged_files(project_root):
    try:
        res = subprocess.run("git diff --cached --name-only", shell=True, capture_output=True, text=True, cwd=project_root)
        if res.returncode == 0:
            return [line.strip() for line in res.stdout.splitlines() if line.strip()]
    except Exception:
        pass
    return []

def determine_max_phase(staged_files):
    if not staged_files:
        return 4

    has_walkthrough = any(f.endswith("walkthrough.md") for f in staged_files)
    has_code = any(not f.endswith(".md") and not f.endswith(".json") and not f.endswith(".yml") and not f.endswith(".yaml") and not f.startswith(".ngpowers") and not f.startswith(".agents") for f in staged_files)
    has_plan = any(f.endswith("task.md") for f in staged_files)
    has_prd = any(f.endswith("prd.md") or f.startswith("docs/") for f in staged_files)
    has_purpose = any(f.endswith("purpose.md") for f in staged_files)

    if has_walkthrough:
        return 4
    if has_code:
        return 3
    if has_plan:
        return 2
    if has_prd:
        return 1
    if has_purpose:
        return 0

    return 4  # Default fallback

def check_task_list(task_path, max_phase):
    if not os.path.exists(task_path):
        return True, []

    uncompleted = []
    current_phase = -1

    with open(task_path, 'r', encoding='utf-8') as f:
        for line_num, line in enumerate(f, 1):
            line_strip = line.strip()
            
            # Detect phase header transitions
            if "Phase 0" in line_strip:
                current_phase = 0
            elif "Phase 1" in line_strip:
                current_phase = 1
            elif "Phase 2" in line_strip:
                current_phase = 2
            elif "Phase 3" in line_strip:
                current_phase = 3
            elif "Phase 4" in line_strip or "Phase 5" in line_strip or "Phase 9" in line_strip:
                current_phase = 4

            if current_phase == -1:
                continue

            # If this line is for a future phase beyond what is currently staged, skip validation
            if current_phase > max_phase:
                continue

            # Check for uncompleted task markers: [ ] or [/]
            if "- [ ]" in line_strip or "- [/]" in line_strip or "- `[ ]`" in line_strip or "- `[/]`" in line_strip:
                if "uncompleted tasks" in line_strip or "in progress tasks" in line_strip:
                    continue
                if "Phase " in line_strip and "- [/]" in line_strip or "Phase " in line_strip and "- `[/]`" in line_strip:
                    continue
                
                task_desc = line_strip.replace("-", "").replace("`", "").replace("[ ]", "").replace("[/]", "").strip()
                uncompleted.append((line_num, task_desc))

    return len(uncompleted) == 0, uncompleted

def check_walkthrough(walkthrough_path):
    if not os.path.exists(walkthrough_path):
        return False, "walkthrough.md is missing. Verification summary is mandatory before committing."

    with open(walkthrough_path, 'r', encoding='utf-8') as f:
        content = f.read().strip()
        
    if len(content) < 100 or "TODO" in content:
        return False, "walkthrough.md is incomplete or contains TODO placeholders."

    return True, ""

def find_git_root():
    curr = os.getcwd()
    while curr != os.path.dirname(curr):
        if (os.path.exists(os.path.join(curr, ".ngpowers")) or 
            os.path.exists(os.path.join(curr, ".agents")) or 
            os.path.exists(os.path.join(curr, "service_spec.md")) or 
            os.path.exists(os.path.join(curr, ".git"))):
            return curr
        curr = os.path.dirname(curr)
    return os.getcwd()

def main():
    script_dir = os.path.dirname(os.path.abspath(__file__))
    parent_dir_name = os.path.basename(os.path.abspath(os.path.join(script_dir, "..")))
    if parent_dir_name.lower() in ("ngpowers", ".ngpowers"):
        parent_dir_name = ".ngpowers"
    elif parent_dir_name.lower() in ("agents", ".agents"):
        parent_dir_name = ".agents"
    
    # Dynamic Git root resolution (agnostic to global or local runs)
    project_root = find_git_root()
    db_path = os.path.join(project_root, parent_dir_name, "state.db")
    
    staged_files = get_staged_files(project_root)
    max_phase = determine_max_phase(staged_files)

    print(f"[{parent_dir_name.replace('.', '').capitalize()} Gate] Context-aware validation active. Max Phase checked: {max_phase}")

    # 1. Database-backed validation (Defects & Vulnerabilities)
    if os.path.exists(db_path):
        try:
            from load_config import get_target_branch
            target_branch = get_target_branch(project_root)
            branch_res = subprocess.run("git rev-parse --abbrev-ref HEAD", shell=True, capture_output=True, text=True, cwd=project_root)
            branch_name = branch_res.stdout.strip() if branch_res.returncode == 0 else target_branch
            
            conn = sqlite3.connect(db_path)
            cursor = conn.cursor()
            
            # Query task matching current branch
            cursor.execute("SELECT task_id, current_phase FROM tasks WHERE branch_name = ? LIMIT 1;", (branch_name,))
            row = cursor.fetchone()
            if row:
                task_id, task_phase = row
                # Override max_phase validation threshold if DB specifies a higher phase
                if task_phase > max_phase:
                    max_phase = task_phase
                    print(f"[{parent_dir_name.replace('.', '').capitalize()} Gate] DB Task {task_id} phase overrides max checked to: {max_phase}")
                
                # Check for open defects (QA)
                cursor.execute("SELECT file_path, line_number, description FROM defects WHERE task_id = ? AND status = 'open';", (task_id,))
                open_defects = cursor.fetchall()
                if open_defects:
                    print(f"\n[{parent_dir_name.replace('.', '').capitalize()} Gate] ERROR: Outstanding QA defects found in database for task {task_id}:", file=sys.stderr)
                    for file_path, line, desc in open_defects:
                        print(f"  - {file_path}:{line} -> {desc}", file=sys.stderr)
                    print(f"\n[{parent_dir_name.replace('.', '').capitalize()} Gate] COMMIT BLOCKED. Resolve QA defects in SQLite database first.\n", file=sys.stderr)
                    conn.close()
                    sys.exit(1)
                
                # Check for open vulnerabilities (Security)
                cursor.execute("SELECT severity, file_path, line_number, description FROM vulnerabilities WHERE task_id = ? AND status = 'open';", (task_id,))
                open_vulns = cursor.fetchall()
                if open_vulns:
                    print(f"\n[{parent_dir_name.replace('.', '').capitalize()} Gate] ERROR: Outstanding security vulnerabilities found in database for task {task_id}:", file=sys.stderr)
                    for severity, file_path, line, desc in open_vulns:
                        print(f"  - [{severity.upper()}] {file_path}:{line} -> {desc}", file=sys.stderr)
                    print(f"\n[{parent_dir_name.replace('.', '').capitalize()} Gate] COMMIT BLOCKED. Address security vulnerabilities in SQLite database first.\n", file=sys.stderr)
                    conn.close()
                    sys.exit(1)
                    
            conn.close()
        except Exception as e:
            print(f"[{parent_dir_name.replace('.', '').capitalize()} Gate] Warning: Failed to query state.db: {e}", file=sys.stderr)

    # 2. Markdown Checklist Fallback Heuristic
    # Dynamic task list detection (stage-driven or root-fallback)
    task_candidates = [os.path.join(project_root, f) for f in staged_files if f.endswith("task.md")]
    task_candidates.append(os.path.join(project_root, "task.md"))
    
    task_paths = list(set([p for p in task_candidates if os.path.exists(p)]))

    task_ok = True
    found_tasks = False
    for path in task_paths:
        found_tasks = True
        ok, uncompleted = check_task_list(path, max_phase)
        if not ok:
            task_ok = False
            print(f"\n[{parent_dir_name.replace('.', '').capitalize()} Gate] ERROR: Uncompleted tasks found in {os.path.relpath(path, project_root)} for Phase <= {max_phase}:", file=sys.stderr)
            for line_num, task in uncompleted:
                print(f"  Line {line_num}: {task}", file=sys.stderr)

    # 3. Walkthrough Check (only required if walkthrough is staged, or if we are at Phase 4 / release stage)
    walkthrough_ok = True
    if found_tasks and max_phase == 4:
        walkthrough_candidates = [os.path.join(project_root, f) for f in staged_files if f.endswith("walkthrough.md")]
        walkthrough_candidates.append(os.path.join(project_root, "walkthrough.md"))
        walkthrough_paths = list(set([p for p in walkthrough_candidates if os.path.exists(p)]))

        for path in walkthrough_paths:
            ok, err_msg = check_walkthrough(path)
            if not ok:
                walkthrough_ok = False
                print(f"\n[{parent_dir_name.replace('.', '').capitalize()} Gate] ERROR in {os.path.relpath(path, project_root)}: {err_msg}", file=sys.stderr)

    if not task_ok or not walkthrough_ok:
        print(f"\n[{parent_dir_name.replace('.', '').capitalize()} Gate] COMMIT BLOCKED. Please complete all tasks for the staged phase and describe verification in walkthrough.md if releasing.", file=sys.stderr)
        print(f"[{parent_dir_name.replace('.', '').capitalize()} Gate] (To bypass this hook in emergency, run: git commit --no-verify)\n", file=sys.stderr)
        sys.exit(1)

    print(f"[{parent_dir_name.replace('.', '').capitalize()} Gate] SDLC quality checks passed successfully!")
    sys.exit(0)

if __name__ == "__main__":
    main()