resolutionflow/backend/app/core/tree_validation.py

"""Tree validation helper module for draft/published workflow."""
from typing import Any

PROCEDURAL_TREE_TYPES = {"procedural", "maintenance"}


class TreeValidationError(Exception):
    """Custom exception for tree validation errors."""
    def __init__(self, field: str, message: str):
        self.field = field
        self.message = message
        super().__init__(f"{field}: {message}")


# --- Troubleshooting Tree Validation ---

def validate_tree_structure(tree_structure: dict[str, Any]) -> tuple[bool, list[dict[str, str]]]:
    """Validate troubleshooting tree structure for publishing.

    A valid tree for publishing must have:
    - A root node with id, type, and appropriate content fields
    - All decision nodes must have a question field
    - All decision nodes with children must have at least 2 children
    - All action nodes must have an action field
    - All solution nodes must have a solution field
    - No orphaned nodes (all nodes reachable from root)

    Args:
        tree_structure: The tree structure dict to validate

    Returns:
        Tuple of (is_valid, list of errors)
        Each error is a dict with 'field' and 'message' keys
    """
    errors = []

    # Check root node exists
    if not tree_structure:
        errors.append({"field": "tree_structure", "message": "Tree structure cannot be empty"})
        return False, errors

    if "id" not in tree_structure:
        errors.append({"field": "tree_structure.id", "message": "Root node must have an id"})

    if "type" not in tree_structure:
        errors.append({"field": "tree_structure.type", "message": "Root node must have a type"})
        return False, errors

    # Validate root node based on type
    _validate_node(tree_structure, "root", errors)

    # Validate all child nodes recursively
    if "children" in tree_structure:
        _validate_children(tree_structure["children"], "root.children", errors)

    # Block publish if any answer placeholder nodes remain
    if _has_answer_nodes(tree_structure):
        errors.append({
            "field": "tree_structure",
            "message": "Answer placeholders must be resolved to a node type before publishing."
        })

    return len(errors) == 0, errors


def _validate_node(node: dict[str, Any], path: str, errors: list[dict[str, str]]) -> None:
    """Validate a single node in the tree structure."""
    node_type = node.get("type")

    if node_type == "decision":
        if "question" not in node or not node["question"]:
            errors.append({
                "field": f"{path}.question",
                "message": "Decision nodes must have a non-empty question"
            })

        # If node has children, must have at least 2 (for decision branches)
        children = node.get("children", [])
        if children and len(children) < 2:
            errors.append({
                "field": f"{path}.children",
                "message": "Decision nodes with children must have at least 2 branches"
            })

    elif node_type == "action":
        if "action" not in node or not node["action"]:
            errors.append({
                "field": f"{path}.action",
                "message": "Action nodes must have a non-empty action"
            })

    elif node_type == "solution":
        if "solution" not in node or not node["solution"]:
            errors.append({
                "field": f"{path}.solution",
                "message": "Solution nodes must have a non-empty solution"
            })

    elif node_type == "answer":
        # Answer nodes are draft-only placeholders — no structural validation needed
        pass

    else:
        errors.append({
            "field": f"{path}.type",
            "message": f"Unknown node type: {node_type}"
        })


def _validate_children(children: list[dict[str, Any]], path: str, errors: list[dict[str, str]]) -> None:
    """Recursively validate child nodes."""
    for i, child in enumerate(children):
        child_path = f"{path}[{i}]"

        if "id" not in child:
            errors.append({"field": f"{child_path}.id", "message": "Child node must have an id"})

        if "type" not in child:
            errors.append({"field": f"{child_path}.type", "message": "Child node must have a type"})
            continue

        _validate_node(child, child_path, errors)

        # Recursively validate grandchildren
        if "children" in child:
            _validate_children(child["children"], f"{child_path}.children", errors)


def _has_answer_nodes(node: dict[str, Any]) -> bool:
    """Recursively check if any node in the tree has type 'answer'."""
    if node.get("type") == "answer":
        return True
    for child in node.get("children", []):
        if _has_answer_nodes(child):
            return True
    return False


# --- Procedural Tree Validation ---

VALID_STEP_TYPES = {"procedure_step", "procedure_end", "section_header"}
VALID_CONTENT_TYPES = {"action", "informational", "verification", "warning"}


def validate_procedural_structure(tree_structure: dict[str, Any]) -> tuple[bool, list[dict[str, str]]]:
    """Validate procedural tree structure for publishing.

    Procedural trees store steps as a flat ordered array in tree_structure["steps"].

    Rules:
    - Must have a non-empty "steps" array
    - Each step must have: id, type, title
    - Only procedure_step and procedure_end types allowed
    - Must have exactly one procedure_end (as the last step)
    - All other steps must be procedure_step
    - No duplicate step IDs
    - Steps with content_type must use valid values

    Args:
        tree_structure: Dict with a "steps" key containing the ordered step array

    Returns:
        Tuple of (is_valid, list of errors)
    """
    errors = []

    if not tree_structure:
        errors.append({"field": "tree_structure", "message": "Tree structure cannot be empty"})
        return False, errors

    steps = tree_structure.get("steps")
    if not steps or not isinstance(steps, list):
        errors.append({"field": "tree_structure.steps", "message": "Procedural tree must have a non-empty steps array"})
        return False, errors

    # Track IDs for uniqueness
    seen_ids: set[str] = set()
    end_count = 0

    for i, step in enumerate(steps):
        path = f"steps[{i}]"

        # Required fields
        step_id = step.get("id")
        if not step_id:
            errors.append({"field": f"{path}.id", "message": "Step must have an id"})
        elif step_id in seen_ids:
            errors.append({"field": f"{path}.id", "message": f"Duplicate step id: {step_id}"})
        else:
            seen_ids.add(step_id)

        step_type = step.get("type")
        if not step_type:
            errors.append({"field": f"{path}.type", "message": "Step must have a type"})
        elif step_type not in VALID_STEP_TYPES:
            errors.append({"field": f"{path}.type", "message": f"Invalid step type: {step_type}. Must be one of: {', '.join(VALID_STEP_TYPES)}"})
        elif step_type == "procedure_end":
            end_count += 1
            # procedure_end must be last step
            if i != len(steps) - 1:
                errors.append({"field": f"{path}.type", "message": "procedure_end must be the last step"})

        if not step.get("title"):
            errors.append({"field": f"{path}.title", "message": "Step must have a non-empty title"})

        # Validate content_type if present
        content_type = step.get("content_type")
        if content_type and content_type not in VALID_CONTENT_TYPES:
            errors.append({"field": f"{path}.content_type", "message": f"Invalid content_type: {content_type}. Must be one of: {', '.join(VALID_CONTENT_TYPES)}"})

    # Must have exactly one end step
    if end_count == 0:
        errors.append({"field": "tree_structure.steps", "message": "Procedural tree must have a procedure_end step as the last step"})
    elif end_count > 1:
        errors.append({"field": "tree_structure.steps", "message": "Procedural tree must have exactly one procedure_end step"})

    return len(errors) == 0, errors


# --- Dispatch ---

def can_publish_tree(
    tree_structure: dict[str, Any],
    name: str,
    description: str | None = None,
    tree_type: str = "troubleshooting",
    intake_form: list[dict[str, Any]] | None = None,
) -> tuple[bool, list[dict[str, str]]]:
    """Check if a tree can be published.

    Dispatches to the appropriate validator based on tree_type.

    Args:
        tree_structure: The tree structure to validate
        name: The tree name
        description: Optional tree description
        tree_type: 'troubleshooting' or 'procedural'
        intake_form: Optional intake form fields (procedural only)

    Returns:
        Tuple of (can_publish, list of errors)
    """
    errors = []

    # Validate name
    if not name or not name.strip():
        errors.append({"field": "name", "message": "Tree must have a name to be published"})

    # Validate structure based on tree type
    if tree_type in PROCEDURAL_TREE_TYPES:
        structure_valid, structure_errors = validate_procedural_structure(tree_structure)
    else:
        structure_valid, structure_errors = validate_tree_structure(tree_structure)
    errors.extend(structure_errors)

    # Validate intake form if present (procedural only)
    if intake_form and tree_type in PROCEDURAL_TREE_TYPES:
        form_valid, form_errors = _validate_intake_form(intake_form)
        errors.extend(form_errors)

    return len(errors) == 0, errors


def _validate_intake_form(intake_form: list[dict[str, Any]]) -> tuple[bool, list[dict[str, str]]]:
    """Validate intake form field definitions."""
    errors = []
    variable_names: set[str] = set()

    for i, field in enumerate(intake_form):
        path = f"intake_form[{i}]"

        var_name = field.get("variable_name")
        if not var_name:
            errors.append({"field": f"{path}.variable_name", "message": "Field must have a variable_name"})
        elif var_name in variable_names:
            errors.append({"field": f"{path}.variable_name", "message": f"Duplicate variable_name: {var_name}"})
        else:
            variable_names.add(var_name)

        if not field.get("label"):
            errors.append({"field": f"{path}.label", "message": "Field must have a label"})

        field_type = field.get("field_type")
        if field_type in ("select", "multi_select"):
            options = field.get("options")
            if not options or len(options) == 0:
                errors.append({"field": f"{path}.options", "message": f"{field_type} fields must have at least one option"})

    return len(errors) == 0, errors