{
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/mcp-servers\/mcp-tasks\/validators\/workflow_validator.py",
"content": "\"\"\"Workflow Validator - Status-Übergangs-Regeln und Pflichtfelder\"\"\"\nfrom typing import Tuple, Optional, Dict, Any\nimport sys\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-tasks\")\nfrom domain.contracts import TaskStatus, TaskType, ExecutorType\n\n\nclass WorkflowValidator:\n \"\"\"Validiert Workflow-Regeln für Tasks\"\"\"\n\n # Gültige Status-Übergänge\n VALID_TRANSITIONS: Dict[str, list] = {\n \"pending\": [\"in_progress\", \"cancelled\"],\n \"in_progress\": [\"completed\", \"failed\", \"pending\", \"cancelled\"],\n \"completed\": [\"pending\"], # Reopen\n \"failed\": [\"pending\", \"in_progress\"],\n \"cancelled\": [\"pending\"], # Reactivate\n }\n\n # Pflichtfelder bei Completion je nach Task-Typ\n COMPLETION_REQUIREMENTS: Dict[str, list] = {\n \"ai_task\": [\"has_result\"],\n \"human_task\": [\"has_result\"],\n \"mixed\": [\"has_result\"],\n }\n\n @staticmethod\n def validate_status_transition(\n current_status: str,\n new_status: str\n ) -> Tuple[bool, str]:\n \"\"\"\n Validiert ob ein Status-Übergang erlaubt ist.\n\n Args:\n current_status: Aktueller Status\n new_status: Gewünschter neuer Status\n\n Returns:\n (is_valid, error_message)\n \"\"\"\n # Normalisiere Status-Werte\n current = current_status.lower() if current_status else \"pending\"\n new = new_status.lower() if new_status else \"\"\n\n # Prüfe ob neuer Status gültig ist\n valid_statuses = [s.value for s in TaskStatus]\n if new not in valid_statuses:\n return False, f\"Ungültiger Status: '{new}'. Erlaubt: {', '.join(valid_statuses)}\"\n\n # Gleicher Status ist immer erlaubt (no-op)\n if current == new:\n return True, \"\"\n\n # Prüfe Übergang\n allowed = WorkflowValidator.VALID_TRANSITIONS.get(current, [])\n if new not in allowed:\n return False, (\n f\"Ungültiger Übergang: {current} → {new}. \"\n f\"Erlaubt von '{current}': {', '.join(allowed)}\"\n )\n\n return True, \"\"\n\n @staticmethod\n def validate_completion(\n task_type: str,\n has_result: bool,\n has_assignment: bool = False\n ) -> Tuple[bool, str]:\n \"\"\"\n Validiert ob ein Task abgeschlossen werden darf.\n\n Args:\n task_type: Typ des Tasks (ai_task, human_task, mixed)\n has_result: Hat der Task mindestens ein Ergebnis?\n has_assignment: Hat der Task mindestens eine Zuweisung?\n\n Returns:\n (is_valid, error_message)\n \"\"\"\n requirements = WorkflowValidator.COMPLETION_REQUIREMENTS.get(\n task_type.lower(), [\"has_result\"]\n )\n\n missing = []\n if \"has_result\" in requirements and not has_result:\n missing.append(\"Ergebnis (tasks_result)\")\n if \"has_assignment\" in requirements and not has_assignment:\n missing.append(\"Zuweisung (tasks_assign)\")\n\n if missing:\n return False, (\n f\"Task kann nicht abgeschlossen werden. \"\n f\"Fehlend: {', '.join(missing)}\"\n )\n\n return True, \"\"\n\n @staticmethod\n def validate_assignment(\n task_type: str,\n assignee_type: str,\n model_name: Optional[str] = None\n ) -> Tuple[bool, str]:\n \"\"\"\n Validiert eine Task-Zuweisung.\n\n Args:\n task_type: Typ des Tasks\n assignee_type: Typ des Zuweisungsempfängers\n model_name: Modellname (bei KI-Zuweisungen)\n\n Returns:\n (is_valid, error_message)\n \"\"\"\n # Prüfe ob assignee_type gültig ist\n valid_types = [e.value for e in ExecutorType]\n if assignee_type.lower() not in valid_types:\n return False, (\n f\"Ungültiger assignee_type: '{assignee_type}'. \"\n f\"Erlaubt: {', '.join(valid_types)}\"\n )\n\n # KI-Zuweisungen brauchen model_name\n ai_types = [\"ollama\", \"claude\", \"anthropic_api\"]\n if assignee_type.lower() in ai_types and not model_name:\n return False, (\n f\"Bei assignee_type='{assignee_type}' ist model_name erforderlich\"\n )\n\n # Warnung bei mismatch (kein Fehler, nur Info)\n warnings = []\n if task_type.lower() == \"human_task\" and assignee_type.lower() in ai_types:\n warnings.append(\n f\"Hinweis: Task ist '{task_type}', wird aber an KI zugewiesen\"\n )\n\n return True, \"; \".join(warnings) if warnings else \"\"\n\n @staticmethod\n def get_allowed_transitions(current_status: str) -> list:\n \"\"\"\n Gibt alle erlaubten Übergänge für einen Status zurück.\n\n Args:\n current_status: Aktueller Status\n\n Returns:\n Liste der erlaubten Ziel-Status\n \"\"\"\n current = current_status.lower() if current_status else \"pending\"\n return WorkflowValidator.VALID_TRANSITIONS.get(current, [])\n\n @staticmethod\n def is_code_task(title: str, description: Optional[str] = None) -> bool:\n \"\"\"\n Prüft ob ein Task ein Code-Task ist (basierend auf Keywords).\n\n Args:\n title: Task-Titel\n description: Task-Beschreibung\n\n Returns:\n True wenn Code-Task erkannt\n \"\"\"\n code_keywords = [\n \"code\", \"implement\", \"fix\", \"bug\", \"feature\", \"refactor\",\n \"php\", \"python\", \"javascript\", \"css\", \"html\",\n \"function\", \"class\", \"method\", \"api\", \"endpoint\",\n \"test\", \"unit\", \"integration\",\n \"migration\", \"database\", \"schema\",\n ]\n\n text = (title + \" \" + (description or \"\")).lower()\n return any(keyword in text for keyword in code_keywords)\n\n @staticmethod\n def validate_quality_gate(\n results: list,\n is_code_task: bool = False,\n ) -> Tuple[bool, str, bool]:\n \"\"\"\n Validiert das Quality Gate vor Task-Completion.\n\n Args:\n results: Liste der Task-Results\n is_code_task: Ist dies ein Code-Task?\n\n Returns:\n (is_valid, warning_message, quality_check_found)\n \"\"\"\n if not is_code_task:\n return True, \"\", False\n\n # Suche nach Quality-Check-Ergebnissen in den Results\n quality_check_found = False\n for result in results:\n response = result.response if hasattr(result, 'response') else str(result.get('response', ''))\n if response and any(kw in response.lower() for kw in ['quality_check', 'phpstan', 'passed', 'issues: 0']):\n quality_check_found = True\n break\n\n if not quality_check_found:\n return True, (\n \"Hinweis: Code-Task ohne Quality-Check abgeschlossen. \"\n \"Empfehlung: quality_check() vor Completion ausführen.\"\n ), False\n\n return True, \"\", True\n",
"numLines": 212,
"startLine": 1,
"totalLines": 212
}
}
}