Task-Completion Guard Hook

Pre-Hook der verhindert, dass Tasks als "completed" markiert werden, ohne dass vorher ein Ergebnis via tasks_result() geschrieben wurde.

Problem

Claude ruft tasks_status(id, "completed") auf, ohne vorher tasks_result() zu rufen. Dies führt zu 77+ Fehlern pro Woche mit der Meldung:

Task kann nicht abgeschlossen werden. Fehlend: Ergebnis (tasks_result)

Lösung

Ein PreToolUse-Hook der VOR dem MCP-Call prüft, ob ein Result existiert und bei Fehlen die Aktion blockiert.

Technische Details

1. Input-Format (stdin)

{
  "tool_name": "mcp__mcp-tasks__tasks_status",
  "tool_input": {
    "id": 425,
    "status": "completed"
  }
}

2. Prüflogik

1. Parse JSON von stdin
2. Prüfe: tool_name == "mcp__mcp-tasks__tasks_status"
3. Prüfe: status == "completed"
4. Wenn ja: DB-Abfrage SELECT COUNT(*) FROM task_results WHERE task_id = ?
5. Wenn count == 0: Block mit Exit-Code 2 + stderr-Nachricht

3. Block-Nachricht

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(id, "completed")

4. Konfiguration in settings.json

{
  "PreToolUse": [
    {
      "matcher": "mcp__mcp-tasks__tasks_status",
      "hooks": [
        {
          "type": "command",
          "command": "/var/www/tools/ki-protokoll/claude-hook/task_completion_guard.py",
          "timeout": 5
        }
      ]
    }
  ]
}

Ablauf

┌─────────────────────────────┐
│ Claude: tasks_status(       │
│   id=425, status="completed"│
│ )                           │
└─────────────┬───────────────┘
              │
              ▼
┌─────────────────────────────┐
│ PreToolUse Hook auslösen    │
│ Matcher: mcp__mcp-tasks__   │
│          tasks_status       │
└─────────────┬───────────────┘
              │
              ▼
┌─────────────────────────────┐
│ task_completion_guard.py    │
│ - Parse Input               │
│ - Check: status=="completed"│
└─────────────┬───────────────┘
              │
       ┌──────┴──────┐
       │             │
       ▼             ▼
┌─────────────┐ ┌─────────────┐
│ status !=   │ │ status ==   │
│ "completed" │ │ "completed" │
└──────┬──────┘ └──────┬──────┘
       │               │
       ▼               ▼
┌─────────────┐ ┌─────────────────┐
│ exit(0)     │ │ DB-Check:       │
│ → Allow     │ │ task_results    │
└─────────────┘ │ WHERE task_id=? │
                └────────┬────────┘
                         │
                  ┌──────┴──────┐
                  │             │
                  ▼             ▼
           ┌──────────┐  ┌──────────┐
           │ count > 0│  │ count = 0│
           └────┬─────┘  └────┬─────┘
                │             │
                ▼             ▼
           ┌─────────┐  ┌─────────────┐
           │ exit(0) │  │ stderr:     │
           │ → Allow │  │ BLOCKIERT   │
           └─────────┘  │ exit(2)     │
                        │ → Block     │
                        └─────────────┘

Implementierungsplan

Schritt 1: Hook-Script erstellen

Datei: /var/www/tools/ki-protokoll/claude-hook/task_completion_guard.py

#!/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()

Schritt 2: Script ausführbar machen

chmod +x /var/www/tools/ki-protokoll/claude-hook/task_completion_guard.py

Schritt 3: settings.json erweitern

Datei: /root/.claude/settings.json

Füge folgenden Eintrag in "PreToolUse" Array hinzu:

{
  "matcher": "mcp__mcp-tasks__tasks_status",
  "hooks": [
    {
      "type": "command",
      "command": "/var/www/tools/ki-protokoll/claude-hook/task_completion_guard.py",
      "timeout": 5
    }
  ]
}

Schritt 4: Testen

# Test 1: Task ohne Result → sollte blockiert werden (Exit 2)
echo '{"tool_name":"mcp__mcp-tasks__tasks_status","tool_input":{"id":999,"status":"completed"}}' | \
  /var/www/tools/ki-protokoll/claude-hook/task_completion_guard.py
echo "Exit code: $?"

# Test 2: Anderer Status → sollte durchgelassen werden (Exit 0)
echo '{"tool_name":"mcp__mcp-tasks__tasks_status","tool_input":{"id":999,"status":"in_progress"}}' | \
  /var/www/tools/ki-protokoll/claude-hook/task_completion_guard.py
echo "Exit code: $?"

# Test 3: Anderes Tool → sollte durchgelassen werden (Exit 0)
echo '{"tool_name":"Bash","tool_input":{"command":"ls"}}' | \
  /var/www/tools/ki-protokoll/claude-hook/task_completion_guard.py
echo "Exit code: $?"

Voraussetzungen

• Python 3.x
• PyMySQL (bereits installiert: pip3 install pymysql)
• python-dotenv (bereits installiert)
• Zugriff auf ki_dev.task_results
• .env Datei im Hook-Verzeichnis mit DB-Credentials

Fail-Safe Verhalten

Bei DB-Verbindungsfehlern wird der Hook "fail-open" - d.h. die Aktion wird durchgelassen, um Claude nicht zu blockieren. Die Server-seitige Validierung in MCP-Tasks greift dann als zweite Verteidigung.

Validierung