from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Optional, Literal, Any, Generic import pandas as pd from pydantic import field_validator, BaseModel, Field, ConfigDict, field_serializer from contracts.CodeLanguages import Language from contracts.GraphAnalyzer import AnalyzeResult, DependencyGraph @dataclass class RepoContext: """Result object carrying all info a coding agent needs.""" remote_url: str branch: str base_branch: str head_commit: str is_new_clone: bool is_new_branch: bool repo_dir: Path = Field(..., description="Absolute path to the cloned repository") original_branch: Optional[str] = None uncommitted_changes: Optional[str] = None # git diff (tracked) untracked_files: list[str] = field(default_factory=list) stash_ref: Optional[str] = None env: dict = field(default_factory=dict) @field_validator("repo_path") @classmethod def repo_must_exist(cls, v: Path) -> Path: if not v.is_dir(): raise ValueError(f"Repository path does not exist or is not a directory: {v}") return v.resolve() class ToolRequirement(BaseModel): """Describes a required external tool and how to install it.""" tool_name: str command: str = Field(..., description="CLI command that must be on PATH") install_guide: str = Field(..., description="Markdown installation instructions") docs_url: Optional[str] = None class AnalysisError(BaseModel): """Structured error returned when analysis cannot proceed.""" error_code: str message: str missing_tools: list[ToolRequirement] = Field(default_factory=list) language: Optional[Language] = None entrypoint: Optional[str] = None class AnalysisInput(BaseModel): """Input schema for the analyzer.""" repo: RepoContext class AnalysisResult(BaseModel): """Top-level result — either a graph or an error.""" model_config = ConfigDict(arbitrary_types_allowed=True) success: bool graph: Optional[DependencyGraph] = None error: Optional[AnalysisError] = None leaf_clusters: Any = None res_all: AnalyzeResult | None = None @field_serializer("leaf_clusters") def _ser_leaf_clusters(self, v: Any, _info): if isinstance(v, pd.DataFrame): return v.to_dict(orient="records") return v # already JSON-safe