273 lines
9 KiB
Python
273 lines
9 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from datetime import datetime
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, List, Dict, Tuple
|
|
|
|
from band_aider.context.memento import Memento
|
|
|
|
if TYPE_CHECKING:
|
|
from band_aider.judgement.file_manager import FileManager
|
|
|
|
|
|
class FileChangeKind(Enum):
|
|
"""FileChangeKind classifies one git-observed file lifecycle change."""
|
|
|
|
Created = "created"
|
|
Deleted = "deleted"
|
|
Moved = "moved"
|
|
Updated = "updated"
|
|
Unchanged = "unchanged"
|
|
Unknown = "unknown"
|
|
|
|
|
|
class ContentType(Enum):
|
|
"""ContentType identifies whether a ClassMeta describes source or test content."""
|
|
|
|
Source = "source"
|
|
Test = "test"
|
|
|
|
|
|
@dataclass
|
|
class ClassMeta:
|
|
"""ClassMeta stores file summary, positive relevance evidence, symbol evidence, and CDC history."""
|
|
|
|
location: str
|
|
summary: str
|
|
relevance: List[str] # relevance contains only positive evidence that the file should be considered for the current task.
|
|
relevant_symbols: Dict[str, str] # Name of symbol (method/attribute) and reason for their relevance.
|
|
cdc: List[Tuple[datetime, str, str]] = field(default_factory=list) # How did the file change, when, who changed it? Why was it changed, what was changed
|
|
content_type: ContentType = ContentType.Source
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Require one explicit, stable source/test ownership declaration."""
|
|
|
|
if not isinstance(self.content_type, ContentType):
|
|
raise ValueError(f"ClassMeta requires a ContentType, got {self.content_type!r}.")
|
|
|
|
def __setstate__(self, state: dict[str, object]) -> None:
|
|
"""Reject checkpoints created before ClassMeta declared its content type."""
|
|
|
|
if "content_type" not in state:
|
|
raise ValueError("Checkpoint ClassMeta is missing content_type; create a new checkpoint.")
|
|
self.__dict__.update(state)
|
|
|
|
def resolve_path(self, file_manager: "FileManager") -> Path:
|
|
"""Resolve this metadata location through its explicit content domain."""
|
|
|
|
if self.content_type is ContentType.Source:
|
|
return file_manager.source_path(self.location)
|
|
if self.content_type is ContentType.Test:
|
|
return file_manager.test_path(self.location)
|
|
raise ValueError(f"Unsupported ClassMeta content type: {self.content_type!r}")
|
|
|
|
def _git_status_name_record(
|
|
self,
|
|
file_manager: "FileManager",
|
|
) -> tuple[str | None, list[str]]:
|
|
"""Return one git status code and repo-relative path payload."""
|
|
|
|
try:
|
|
path = self.resolve_path(file_manager)
|
|
location = file_manager.repo_location(path)
|
|
except (AttributeError, OSError, ValueError):
|
|
location = self.location
|
|
|
|
repo_root = file_manager.repo_root
|
|
if not self._git_work_tree_available(file_manager, repo_root):
|
|
return None, [location]
|
|
|
|
repo_record = self._git_status_record_for(file_manager, repo_root, location, pathspec=None)
|
|
if repo_record is not None:
|
|
return repo_record
|
|
|
|
pathspec_record = self._git_status_record_for(file_manager, repo_root, location, pathspec=location)
|
|
if pathspec_record is not None:
|
|
return pathspec_record
|
|
|
|
return "", [location]
|
|
|
|
def _git_work_tree_available(self, file_manager: "FileManager", repo_root: Path) -> bool:
|
|
"""Return whether git can inspect status under this repository root."""
|
|
|
|
try:
|
|
result = file_manager.repo_exec.run_cli(
|
|
["git", "rev-parse", "--is-inside-work-tree"],
|
|
cwd=repo_root,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
except OSError:
|
|
return False
|
|
return result.returncode == 0 and result.stdout.strip() == "true"
|
|
|
|
def _git_status_record_for(
|
|
self,
|
|
file_manager: "FileManager",
|
|
repo_root: Path,
|
|
location: str,
|
|
*,
|
|
pathspec: str | None,
|
|
) -> tuple[str, list[str]] | None:
|
|
"""Return a matching git status record for one location."""
|
|
|
|
command = ["git", "status", "--porcelain=v1", "-M", "--untracked-files=all"]
|
|
if pathspec is not None:
|
|
command.extend(["--", pathspec])
|
|
try:
|
|
result = file_manager.repo_exec.run_cli(
|
|
command,
|
|
cwd=repo_root,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
except OSError:
|
|
return None
|
|
|
|
if result.returncode != 0:
|
|
return None
|
|
|
|
for line in result.stdout.splitlines():
|
|
record = self._parse_git_status_line(line)
|
|
if record is None:
|
|
continue
|
|
code, paths = record
|
|
if location in paths:
|
|
return code, paths
|
|
return None
|
|
|
|
def _parse_git_status_line(self, line: str) -> tuple[str, list[str]] | None:
|
|
"""Parse one git status v1 line."""
|
|
|
|
if not line:
|
|
return None
|
|
code = line[:2]
|
|
payload = line[3:]
|
|
if " -> " in payload:
|
|
old, new = payload.split(" -> ", 1)
|
|
return code, [old.strip(), new.strip()]
|
|
return code, [payload.strip()]
|
|
|
|
def file_change_kind(
|
|
self,
|
|
file_manager: "FileManager",
|
|
) -> FileChangeKind:
|
|
"""Return the git-observed file lifecycle change kind."""
|
|
|
|
code, paths = self._git_status_name_record(file_manager)
|
|
if code is None:
|
|
return FileChangeKind.Unknown
|
|
if not code:
|
|
return FileChangeKind.Unchanged
|
|
if "R" in code and len(paths) == 2:
|
|
return FileChangeKind.Moved
|
|
if code == "??" or "A" in code:
|
|
return FileChangeKind.Created
|
|
if "D" in code:
|
|
return FileChangeKind.Deleted
|
|
if any(marker in code for marker in ("M", "T", "U")):
|
|
return FileChangeKind.Updated
|
|
return FileChangeKind.Unknown
|
|
|
|
def file_movement(
|
|
self,
|
|
file_manager: "FileManager",
|
|
) -> tuple[str, str]:
|
|
"""Return repo-relative from/to paths for the git-observed file lifecycle change."""
|
|
|
|
code, paths = self._git_status_name_record(file_manager)
|
|
kind = self._file_change_kind_for_status(code, paths)
|
|
if kind is FileChangeKind.Created:
|
|
return "", paths[-1] if paths else self.location
|
|
if kind is FileChangeKind.Deleted:
|
|
return paths[0] if paths else self.location, ""
|
|
if kind is FileChangeKind.Moved and len(paths) == 2:
|
|
return paths[0], paths[1]
|
|
if paths:
|
|
return paths[-1], paths[-1]
|
|
return self.location, self.location
|
|
|
|
def _file_change_kind_for_status(
|
|
self,
|
|
code: str | None,
|
|
paths: list[str],
|
|
) -> FileChangeKind:
|
|
"""Classify one parsed git status code."""
|
|
|
|
if code is None:
|
|
return FileChangeKind.Unknown
|
|
if not code:
|
|
return FileChangeKind.Unchanged
|
|
if "R" in code and len(paths) == 2:
|
|
return FileChangeKind.Moved
|
|
if code == "??" or "A" in code:
|
|
return FileChangeKind.Created
|
|
if "D" in code:
|
|
return FileChangeKind.Deleted
|
|
if any(marker in code for marker in ("M", "T", "U")):
|
|
return FileChangeKind.Updated
|
|
return FileChangeKind.Unknown
|
|
|
|
def get_source(
|
|
self,
|
|
file_manager: "FileManager",
|
|
) -> str:
|
|
"""Return the current file content for this ClassMeta."""
|
|
|
|
try:
|
|
path = self.resolve_path(file_manager)
|
|
except (OSError, ValueError):
|
|
return ""
|
|
try:
|
|
repo_exec = file_manager.repo_exec
|
|
if not repo_exec.exists(path):
|
|
return ""
|
|
return repo_exec.read_text(path, encoding="utf-8")
|
|
except OSError:
|
|
return ""
|
|
|
|
def get_diff(
|
|
self,
|
|
file_manager: "FileManager",
|
|
) -> str:
|
|
"""Return the current git diff for this ClassMeta."""
|
|
|
|
try:
|
|
path = self.resolve_path(file_manager)
|
|
except (OSError, ValueError):
|
|
return ""
|
|
return file_manager.git_diff_for(path)
|
|
|
|
def get_hash(
|
|
self,
|
|
file_manager: "FileManager",
|
|
) -> str:
|
|
"""Return a stable content hash for this ClassMeta."""
|
|
|
|
source = self.get_source(file_manager)
|
|
return hashlib.md5(source.encode("utf-8")).hexdigest()
|
|
|
|
def set_or_get_memento(
|
|
self,
|
|
file_manager: "FileManager",
|
|
*,
|
|
stack: bool = False,
|
|
) -> Memento:
|
|
"""Return the existing file Memento or persist the current file as a new one."""
|
|
|
|
file_path = self.resolve_path(file_manager)
|
|
if stack:
|
|
return Memento.first_available(file_path, file_manager.repo_exec).capture()
|
|
return (
|
|
Memento.current(file_path, file_manager.repo_exec)
|
|
or Memento(file_path, file_manager.repo_exec)
|
|
).capture()
|