workflow_validator.py

Code Hygiene Score: 92

Keine Issues gefunden.

Dependencies 8

Klassen 1

Code

"""Workflow Validator - Status-Übergangs-Regeln und Pflichtfelder"""
from typing import Tuple, Optional, Dict, Any
import sys

sys.path.insert(0, "/var/www/mcp-servers/mcp_tasks")
from domain.contracts import TaskStatus, TaskType, ExecutorType


class WorkflowValidator:
    """Validiert Workflow-Regeln für Tasks"""

    # Gültige Status-Übergänge
    VALID_TRANSITIONS: Dict[str, list] = {
        "pending": ["in_progress", "cancelled"],
        "in_progress": ["completed", "failed", "pending", "cancelled"],
        "completed": ["pending"],  # Reopen
        "failed": ["pending", "in_progress"],
        "cancelled": ["pending"],  # Reactivate
    }

    # Pflichtfelder bei Completion je nach Task-Typ
    COMPLETION_REQUIREMENTS: Dict[str, list] = {
        "ai_task": ["has_result"],
        "human_task": ["has_result"],
        "mixed": ["has_result"],
    }

    @staticmethod
    def validate_status_transition(
        current_status: str,
        new_status: str
    ) -> Tuple[bool, str]:
        """
        Validiert ob ein Status-Übergang erlaubt ist.

        Args:
            current_status: Aktueller Status
            new_status: Gewünschter neuer Status

        Returns:
            (is_valid, error_message)
        """
        # Normalisiere Status-Werte
        current = current_status.lower() if current_status else "pending"
        new = new_status.lower() if new_status else ""

        # Prüfe ob neuer Status gültig ist
        valid_statuses = [s.value for s in TaskStatus]
        if new not in valid_statuses:
            return False, f"Ungültiger Status: '{new}'. Erlaubt: {', '.join(valid_statuses)}"

        # Gleicher Status ist immer erlaubt (no-op)
        if current == new:
            return True, ""

        # Prüfe Übergang
        allowed = WorkflowValidator.VALID_TRANSITIONS.get(current, [])
        if new not in allowed:
            return False, (
                f"Ungültiger Übergang: {current} → {new}. "
                f"Erlaubt von '{current}': {', '.join(allowed)}"
            )

        return True, ""

    @staticmethod
    def validate_completion(
        task_type: str,
        has_result: bool,
        has_assignment: bool = False
    ) -> Tuple[bool, str]:
        """
        Validiert ob ein Task abgeschlossen werden darf.

        Args:
            task_type: Typ des Tasks (ai_task, human_task, mixed)
            has_result: Hat der Task mindestens ein Ergebnis?
            has_assignment: Hat der Task mindestens eine Zuweisung?

        Returns:
            (is_valid, error_message)
        """
        requirements = WorkflowValidator.COMPLETION_REQUIREMENTS.get(
            task_type.lower(), ["has_result"]
        )

        missing = []
        if "has_result" in requirements and not has_result:
            missing.append("Ergebnis (tasks_result)")
        if "has_assignment" in requirements and not has_assignment:
            missing.append("Zuweisung (tasks_assign)")

        if missing:
            return False, (
                f"Task kann nicht abgeschlossen werden. "
                f"Fehlend: {', '.join(missing)}"
            )

        return True, ""

    @staticmethod
    def validate_assignment(
        task_type: str,
        assignee_type: str,
        model_name: Optional[str] = None
    ) -> Tuple[bool, str]:
        """
        Validiert eine Task-Zuweisung.

        Args:
            task_type: Typ des Tasks
            assignee_type: Typ des Zuweisungsempfängers
            model_name: Modellname (bei KI-Zuweisungen)

        Returns:
            (is_valid, error_message)
        """
        # Prüfe ob assignee_type gültig ist
        valid_types = [e.value for e in ExecutorType]
        if assignee_type.lower() not in valid_types:
            return False, (
                f"Ungültiger assignee_type: '{assignee_type}'. "
                f"Erlaubt: {', '.join(valid_types)}"
            )

        # KI-Zuweisungen brauchen model_name
        ai_types = ["ollama", "claude", "anthropic_api"]
        if assignee_type.lower() in ai_types and not model_name:
            return False, (
                f"Bei assignee_type='{assignee_type}' ist model_name erforderlich"
            )

        # Warnung bei mismatch (kein Fehler, nur Info)
        warnings = []
        if task_type.lower() == "human_task" and assignee_type.lower() in ai_types:
            warnings.append(
                f"Hinweis: Task ist '{task_type}', wird aber an KI zugewiesen"
            )

        return True, "; ".join(warnings) if warnings else ""

    @staticmethod
    def get_allowed_transitions(current_status: str) -> list:
        """
        Gibt alle erlaubten Übergänge für einen Status zurück.

        Args:
            current_status: Aktueller Status

        Returns:
            Liste der erlaubten Ziel-Status
        """
        current = current_status.lower() if current_status else "pending"
        return WorkflowValidator.VALID_TRANSITIONS.get(current, [])

    @staticmethod
    def is_code_task(title: str, description: Optional[str] = None) -> bool:
        """
        Prüft ob ein Task ein Code-Task ist (basierend auf Keywords).

        Args:
            title: Task-Titel
            description: Task-Beschreibung

        Returns:
            True wenn Code-Task erkannt
        """
        code_keywords = [
            "code", "implement", "fix", "bug", "feature", "refactor",
            "php", "python", "javascript", "css", "html",
            "function", "class", "method", "api", "endpoint",
            "test", "unit", "integration",
            "migration", "database", "schema",
        ]

        text = (title + " " + (description or "")).lower()
        return any(keyword in text for keyword in code_keywords)

    @staticmethod
    def validate_quality_gate(
        results: list,
        is_code_task: bool = False,
    ) -> Tuple[bool, str, bool]:
        """
        Validiert das Quality Gate vor Task-Completion.

        Args:
            results: Liste der Task-Results
            is_code_task: Ist dies ein Code-Task?

        Returns:
            (is_valid, warning_message, quality_check_found)
        """
        if not is_code_task:
            return True, "", False

        # Suche nach Quality-Check-Ergebnissen in den Results
        quality_check_found = False
        for result in results:
            response = result.response if hasattr(result, 'response') else str(result.get('response', ''))
            if response and any(kw in response.lower() for kw in ['quality_check', 'phpstan', 'passed', 'issues: 0']):
                quality_check_found = True
                break

        if not quality_check_found:
            return True, (
                "Hinweis: Code-Task ohne Quality-Check abgeschlossen. "
                "Empfehlung: quality_check() vor Completion ausführen."
            ), False

        return True, "", True
← Übersicht Graph