"""Sandboxowane narzędzia plikowe - jedyny kanał kontaktu agenta z repozytorium. Model nigdy nie dostaje powłoki. Każda operacja jest: - ograniczona do katalogu repozytorium (brak wyjścia przez `..` i dowiązania), - sprawdzana względem listy zakazanych ścieżek (pipeline, sekrety, manifesty), - limitowana rozmiarem i liczbą zmienionych plików, - zapisywana w śladzie audytowym. Guardraile z instrukcji APM są dla modelu. Ten moduł jest dla audytora. """ from __future__ import annotations import fnmatch import subprocess from pathlib import Path from ..config import Settings from ..observability.audit import AuditLog _TEXT_SUFFIXES = { ".py", ".java", ".kt", ".js", ".ts", ".tsx", ".go", ".rb", ".rs", ".sql", ".toml", ".cfg", ".ini", ".txt", ".md", ".yaml", ".yml", ".json", ".xml", ".gradle", ".properties", ".sh", ".tf", "", } class WorkspaceError(RuntimeError): """Błąd zwracany agentowi jako czytelny komunikat, nie jako wyjątek przerywający przebieg.""" class WorkspaceTools: def __init__(self, root: Path | str, settings: Settings, audit: AuditLog) -> None: self.root = Path(root).resolve() if not self.root.is_dir(): raise WorkspaceError(f"Katalog repozytorium nie istnieje: {self.root}") self.settings = settings self.audit = audit self.changed_files: set[str] = set() # ------------------------------------------------------------------ util def _resolve(self, path: str, for_write: bool = False) -> Path: candidate = (self.root / path).resolve() if not candidate.is_relative_to(self.root): raise WorkspaceError(f"Ścieżka poza repozytorium jest zabroniona: {path}") relative = candidate.relative_to(self.root).as_posix() for pattern in self.settings.deny_globs: if fnmatch.fnmatch(relative, pattern) or fnmatch.fnmatch(relative, pattern.replace("**/", "")): raise WorkspaceError( f"Ścieżka '{relative}' jest objęta zakazem modyfikacji (guardrail: {pattern}). " "Zgłoś potrzebę zmiany jako requires_human." ) if for_write and len(self.changed_files | {relative}) > self.settings.max_files_changed: raise WorkspaceError( f"Przekroczony budżet zmienionych plików ({self.settings.max_files_changed}). " "Zakres zmiany jest zbyt szeroki - zatrzymaj się i zgłoś to w podsumowaniu." ) return candidate # ----------------------------------------------------------------- tools def list_files(self, subdirectory: str = ".", pattern: str = "*") -> str: """Wypisuje pliki repozytorium. Użyj do rozpoznania struktury projektu. Args: subdirectory: katalog względem korzenia repozytorium (domyślnie cały projekt). pattern: wzorzec glob nazwy pliku, np. '*.py'. """ base = self._resolve(subdirectory) results: list[str] = [] for path in sorted(base.rglob(pattern)): if not path.is_file() or any( part in {".git", "__pycache__", ".venv", "node_modules"} for part in path.parts ): continue results.append(path.relative_to(self.root).as_posix()) if len(results) >= 500: results.append("... (lista obcięta do 500 pozycji)") break self.audit.record( "list_files", {"subdirectory": subdirectory, "pattern": pattern}, detail=f"{len(results)} plików" ) return "\n".join(results) or "(brak plików)" def read_file(self, path: str) -> str: """Zwraca zawartość pliku z numerami linii. Zawsze czytaj plik przed jego edycją. Args: path: ścieżka względem korzenia repozytorium. """ target = self._resolve(path) if not target.is_file(): self.audit.record("read_file", {"path": path}, ok=False, detail="brak pliku") raise WorkspaceError(f"Plik nie istnieje: {path}") if target.stat().st_size > self.settings.max_file_bytes: raise WorkspaceError(f"Plik {path} przekracza limit {self.settings.max_file_bytes} bajtów") content = target.read_text(encoding="utf-8", errors="replace") self.audit.record("read_file", {"path": path}, detail=f"{len(content)} znaków") numbered = "\n".join(f"{i:>4}| {line}" for i, line in enumerate(content.splitlines(), start=1)) return numbered or "(plik pusty)" def search_repo(self, pattern: str, file_glob: str = "*") -> str: """Wyszukuje wzorzec (regex) w repozytorium i zwraca dopasowania z numerami linii. Args: pattern: wyrażenie regularne, np. 'from acme import'. file_glob: ograniczenie do typu plików, np. '*.py'. """ command = [ "grep", "-rniE", "--line-number", f"--include={file_glob}", "--exclude-dir=.git", "--exclude-dir=__pycache__", "--exclude-dir=.venv", "--exclude-dir=node_modules", pattern, ".", ] proc = subprocess.run(command, cwd=self.root, capture_output=True, text=True, timeout=60) output = proc.stdout.strip() lines = output.splitlines()[:200] self.audit.record("search_repo", {"pattern": pattern, "file_glob": file_glob}, detail=f"{len(lines)} dopasowań") return "\n".join(lines) or "(brak dopasowań)" def replace_in_file(self, path: str, old_text: str, new_text: str) -> str: """Zastępuje dokładnie jedno wystąpienie fragmentu w pliku. Podstawowe narzędzie edycji. Fragment `old_text` musi być unikalny w pliku i przepisany co do znaku (wraz z wcięciami). Jeśli fragment występuje wielokrotnie - poszerz go o sąsiednie linie. Args: path: ścieżka względem korzenia repozytorium. old_text: dokładny fragment do zastąpienia. new_text: nowa treść fragmentu. """ target = self._resolve(path, for_write=True) if not target.is_file(): raise WorkspaceError(f"Plik nie istnieje: {path}") content = target.read_text(encoding="utf-8") occurrences = content.count(old_text) if occurrences == 0: self.audit.record("replace_in_file", {"path": path}, ok=False, detail="brak dopasowania") raise WorkspaceError( f"Nie znaleziono podanego fragmentu w {path}. Odczytaj plik ponownie i przepisz fragment dokładnie." ) if occurrences > 1: self.audit.record("replace_in_file", {"path": path}, ok=False, detail=f"{occurrences} dopasowań") raise WorkspaceError( f"Fragment występuje {occurrences} razy w {path}. Poszerz go o sąsiednie linie, aby był jednoznaczny." ) target.write_text(content.replace(old_text, new_text, 1), encoding="utf-8") relative = target.relative_to(self.root).as_posix() self.changed_files.add(relative) self.audit.record("replace_in_file", {"path": path}, detail="ok") return f"Zmieniono {relative}." def write_file(self, path: str, content: str) -> str: """Zapisuje plik w całości. Używaj wyłącznie dla plików nowych - do edycji służy replace_in_file. Args: path: ścieżka względem korzenia repozytorium. content: pełna treść pliku. """ target = self._resolve(path, for_write=True) if target.suffix not in _TEXT_SUFFIXES: raise WorkspaceError(f"Niedozwolony typ pliku do zapisu: {target.suffix}") if len(content.encode("utf-8")) > self.settings.max_file_bytes: raise WorkspaceError("Treść przekracza limit rozmiaru pliku") target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") relative = target.relative_to(self.root).as_posix() self.changed_files.add(relative) self.audit.record("write_file", {"path": path}, detail=f"{len(content)} znaków") return f"Zapisano {relative}." def get_diff(self) -> str: """Zwraca aktualny diff repozytorium (git diff wraz z plikami nieśledzonymi).""" subprocess.run(["git", "add", "-AN"], cwd=self.root, capture_output=True, text=True) proc = subprocess.run(["git", "diff"], cwd=self.root, capture_output=True, text=True, timeout=60) diff = proc.stdout self.audit.record("get_diff", {}, detail=f"{len(diff)} znaków") if not diff.strip(): return "(brak zmian w repozytorium)" return diff[:60_000]