76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
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 |