task_completion_guard.py

• Pfad: /var/www/tools/ki-protokoll/claude-hook/task_completion_guard.py
• Namespace: claude-hook
• Zeilen: 96 | Größe: 2,523 Bytes
• Geändert: 2025-12-27 10:24:20 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 98

• Dependencies: 90 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 100 (10%)

Keine Issues gefunden.

Dependencies 6

• use json
• use sys
• use os
• use pathlib.Path
• use dotenv.load_dotenv
• use pymysql

Funktionen 2

• check_result_exists() Zeile 46
• main() Zeile 64

Code

#!/usr/bin/env python3
"""
Task Completion Guard Hook

Blockiert tasks_status(completed) wenn kein Result existiert.
Erzwingt korrekten Workflow: tasks_result() vor tasks_status(completed).

Trigger: PreToolUse (mcp__mcp-tasks__tasks_status)
"""

import json
import sys
import os
from pathlib import Path
from dotenv import load_dotenv
import pymysql

# Lade Environment aus .env im Hook-Verzeichnis
load_dotenv(Path(__file__).parent / '.env')

DB_CONFIG = {
    'host': os.environ.get('CLAUDE_DB_HOST', 'localhost'),
    'port': int(os.environ.get('CLAUDE_DB_PORT', '3306')),
    'user': os.environ.get('CLAUDE_DB_USER', 'root'),
    'password': os.environ.get('CLAUDE_DB_PASSWORD', ''),
    'database': 'ki_dev',
    'charset': 'utf8mb4'
}

BLOCK_MESSAGE = """BLOCKIERT: Task kann nicht als 'completed' markiert werden!

GRUND: Kein Ergebnis vorhanden (tasks_result fehlt).

LÖSUNG: Vor tasks_status(id, "completed") MUSS tasks_result() aufgerufen werden:

    tasks_result(
        id={task_id},
        response="Zusammenfassung der erledigten Arbeit...",
        executor="claude",
        executor_type="claude"
    )

Erst danach: tasks_status({task_id}, "completed")"""


def check_result_exists(task_id: int) -> bool:
    """Prüft ob ein Result für den Task existiert."""
    try:
        conn = pymysql.connect(**DB_CONFIG)
        cursor = conn.cursor()
        cursor.execute(
            "SELECT COUNT(*) FROM task_results WHERE task_id = %s",
            (task_id,)
        )
        count = cursor.fetchone()[0]
        conn.close()
        return count > 0
    except Exception as e:
        # Bei DB-Fehler durchlassen (fail-open)
        print(f"DB-Check fehlgeschlagen: {e}", file=sys.stderr)
        return True


def main():
    try:
        input_data = json.load(sys.stdin)
    except json.JSONDecodeError:
        sys.exit(0)

    # Nur mcp__mcp-tasks__tasks_status prüfen
    tool_name = input_data.get("tool_name", "")
    if tool_name != "mcp__mcp-tasks__tasks_status":
        sys.exit(0)

    # Prüfe ob Status auf "completed" gesetzt wird
    tool_input = input_data.get("tool_input", {})
    status = tool_input.get("status", "")
    task_id = tool_input.get("id")

    if status.lower() != "completed":
        sys.exit(0)

    if not task_id:
        sys.exit(0)

    # Prüfe ob Result existiert
    if not check_result_exists(task_id):
        print(BLOCK_MESSAGE.format(task_id=task_id), file=sys.stderr)
        sys.exit(2)

    sys.exit(0)


if __name__ == "__main__":
    main()