microsoft--agent-framework/python/scripts/sample_validation/models.py

# Copyright (c) Microsoft. All rights reserved.

"""Data models for the sample validation system."""

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from pathlib import Path

from agent_framework import Workflow
from agent_framework.github import GitHubCopilotAgent


@dataclass
class ValidationConfig:
    """Configuration for the validation workflow."""

    samples_dir: Path
    python_root: Path
    subdir: str | None = None
    exclude: list[str] | None = None
    max_parallel_workers: int = 10
    playbooks_dir: Path | None = None
    use_cache: bool = True
    agent_timeout: int = 120


@dataclass
class SampleInfo:
    """Information about a discovered sample file."""

    path: Path
    relative_path: str

    @classmethod
    def from_path(cls, path: Path, samples_dir: Path) -> "SampleInfo":
        """Create SampleInfo from a file path."""
        return cls(
            path=path,
            relative_path=str(path.relative_to(samples_dir)),
        )


@dataclass
class DiscoveryResult:
    """Result of sample discovery."""

    samples: list[SampleInfo]


@dataclass
class ReplayResult:
    """Outcome of attempting to replay cached playbooks for discovered samples.

    Samples whose cached playbook replayed successfully are captured in
    ``cached_results`` and skip the agent entirely. Everything else flows to the
    agent-driven validation as ``remaining_samples``.
    """

    remaining_samples: list[SampleInfo]
    cached_results: list["RunResult"] = field(default_factory=list)  # type: ignore[assignment]


@dataclass
class WorkflowCreationResult:
    """Result of creating a nested per-sample concurrent workflow."""

    samples: list[SampleInfo]
    workflow: Workflow | None
    agents: list[GitHubCopilotAgent]
    cached_results: list["RunResult"] = field(default_factory=list)  # type: ignore[assignment]


class RunStatus(Enum):
    """Status of a sample run."""

    SUCCESS = "success"
    FAILURE = "failure"
    MISSING_SETUP = "missing_setup"


@dataclass
class RunResult:
    """Result of running a single sample."""

    sample: SampleInfo
    status: RunStatus
    output: str
    error: str


@dataclass
class ExecutionResult:
    """Result of sample execution."""

    results: list[RunResult]


@dataclass
class Report:
    """Final validation report."""

    timestamp: datetime
    total_samples: int
    success_count: int
    failure_count: int
    missing_setup_count: int
    results: list[RunResult] = field(default_factory=list)  # type: ignore

    def to_markdown(self) -> str:
        """Generate a markdown report."""
        lines = [
            "# Sample Validation Report",
            "",
            f"**Generated:** {self.timestamp.isoformat()}",
            "",
            "## Summary",
            "",
            "| Metric | Count |",
            "|--------|-------|",
            f"| Total Samples | {self.total_samples} |",
            f"| [PASS] Success | {self.success_count} |",
            f"| [FAIL] Failure | {self.failure_count} |",
            f"| [MISSING_SETUP] Missing Setup | {self.missing_setup_count} |",
            "",
            "## Detailed Results",
            "",
        ]

        # Group by status
        for status in [RunStatus.FAILURE, RunStatus.MISSING_SETUP, RunStatus.SUCCESS]:
            status_results = [r for r in self.results if r.status == status]
            if not status_results:
                continue

            status_label = {
                RunStatus.SUCCESS: "[PASS]",
                RunStatus.FAILURE: "[FAIL]",
                RunStatus.MISSING_SETUP: "[MISSING_SETUP]",
            }

            lines.append(f"### {status_label[status]} {status.value.title()} ({len(status_results)})")
            lines.append("")

            for result in status_results:
                lines.append(f"- **{result.sample.relative_path}**")
                if result.error:
                    # Truncate long errors
                    error_preview = result.error[:200] + "..." if len(result.error) > 200 else result.error
                    lines.append(f"  - Error: `{error_preview}`")
            lines.append("")

        return "\n".join(lines)

    def to_dict(self) -> dict[str, object]:
        """Convert report to dictionary for JSON serialization."""
        return {
            "timestamp": self.timestamp.isoformat(),
            "summary": {
                "total_samples": self.total_samples,
                "success_count": self.success_count,
                "failure_count": self.failure_count,
                "missing_setup_count": self.missing_setup_count,
            },
            "results": [
                {
                    "path": r.sample.relative_path,
                    "status": r.status.value,
                    "output": r.output,
                    "error": r.error,
                }
                for r in self.results
            ],
        }