{
"event": "PostToolUse",
"tool_name": "Read",
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/task_completion_guard.py",
"content": "#!\/usr\/bin\/env python3\n\"\"\"\nTask Completion Guard Hook\n\nBlockiert tasks_status(completed) wenn kein Result existiert.\nErzwingt korrekten Workflow: tasks_result() vor tasks_status(completed).\n\nTrigger: PreToolUse (mcp__mcp-tasks__tasks_status)\n\"\"\"\n\nimport json\nimport sys\nimport os\nfrom pathlib import Path\nfrom dotenv import load_dotenv\nimport pymysql\n\n# Lade Environment aus .env im Hook-Verzeichnis\nload_dotenv(Path(__file__).parent \/ '.env')\n\nDB_CONFIG = {\n 'host': os.environ.get('CLAUDE_DB_HOST', 'localhost'),\n 'port': int(os.environ.get('CLAUDE_DB_PORT', '3306')),\n 'user': os.environ.get('CLAUDE_DB_USER', 'root'),\n 'password': os.environ.get('CLAUDE_DB_PASSWORD', ''),\n 'database': 'ki_dev',\n 'charset': 'utf8mb4'\n}\n\nBLOCK_MESSAGE = \"\"\"BLOCKIERT: Task kann nicht als 'completed' markiert werden!\n\nGRUND: Kein Ergebnis vorhanden (tasks_result fehlt).\n\nLÖSUNG: Vor tasks_status(id, \"completed\") MUSS tasks_result() aufgerufen werden:\n\n tasks_result(\n id={task_id},\n response=\"Zusammenfassung der erledigten Arbeit...\",\n executor=\"claude\",\n executor_type=\"claude\"\n )\n\nErst danach: tasks_status({task_id}, \"completed\")\"\"\"\n\n\ndef check_result_exists(task_id: int) -> bool:\n \"\"\"Prüft ob ein Result für den Task existiert.\"\"\"\n try:\n conn = pymysql.connect(**DB_CONFIG)\n cursor = conn.cursor()\n cursor.execute(\n \"SELECT COUNT(*) FROM task_results WHERE task_id = %s\",\n (task_id,)\n )\n count = cursor.fetchone()[0]\n conn.close()\n return count > 0\n except Exception as e:\n # Bei DB-Fehler durchlassen (fail-open)\n print(f\"DB-Check fehlgeschlagen: {e}\", file=sys.stderr)\n return True\n\n\ndef main():\n try:\n input_data = json.load(sys.stdin)\n except json.JSONDecodeError:\n sys.exit(0)\n\n # Nur mcp__mcp-tasks__tasks_status prüfen\n tool_name = input_data.get(\"tool_name\", \"\")\n if tool_name != \"mcp__mcp-tasks__tasks_status\":\n sys.exit(0)\n\n # Prüfe ob Status auf \"completed\" gesetzt wird\n tool_input = input_data.get(\"tool_input\", {})\n status = tool_input.get(\"status\", \"\")\n task_id = tool_input.get(\"id\")\n\n if status.lower() != \"completed\":\n sys.exit(0)\n\n if not task_id:\n sys.exit(0)\n\n # Prüfe ob Result existiert\n if not check_result_exists(task_id):\n print(BLOCK_MESSAGE.format(task_id=task_id), file=sys.stderr)\n sys.exit(2)\n\n sys.exit(0)\n\n\nif __name__ == \"__main__\":\n main()\n",
"numLines": 96,
"startLine": 1,
"totalLines": 96
}
}
}