Spaces:
Sleeping
Sleeping
| import glob as _glob | |
| import os | |
| import subprocess | |
| from pathlib import Path | |
| from typing import Optional | |
| import pandas as pd | |
| from docx import Document | |
| from pypdf import PdfReader | |
| # Agent writes are confined to this directory. Defaults to `.lilith/scratch` | |
| # under the CWD; override with set_write_root() at startup if you want a | |
| # per-run tmpfs-style root. The LLM must not be able to change this. | |
| _WRITE_ROOT: Path = Path(".lilith/scratch").resolve() | |
| # Default directories pruned by find_files. Keeps results useful on real | |
| # codebases by skipping vendored/build/cache trees. | |
| _DEFAULT_EXCLUDES: tuple[str, ...] = ( | |
| "node_modules", | |
| ".git", | |
| ".venv", | |
| "__pycache__", | |
| ".pytest_cache", | |
| ".mypy_cache", | |
| "dist", | |
| "build", | |
| ".tox", | |
| ) | |
| def set_write_root(root: str | Path) -> None: | |
| """Set the sandboxed root directory for write_file. Call once at startup.""" | |
| global _WRITE_ROOT | |
| _WRITE_ROOT = Path(root).resolve() | |
| _WRITE_ROOT.mkdir(parents=True, exist_ok=True) | |
| def _resolve_safe_write_path(path: str) -> Path: | |
| """Resolve `path` against the write root. Raises ValueError on unsafe input. | |
| Rejects absolute paths and any relative path that resolves outside the root | |
| (e.g. via `..`). Returns a fully resolved Path inside the root. | |
| """ | |
| if not isinstance(path, str) or not path: | |
| raise ValueError("path must be a non-empty string") | |
| p = Path(path) | |
| if p.is_absolute(): | |
| raise ValueError(f"absolute paths are not allowed: {path!r}") | |
| root = _WRITE_ROOT.resolve() | |
| candidate = (root / p).resolve() | |
| try: | |
| candidate.relative_to(root) | |
| except ValueError: | |
| raise ValueError(f"path would escape write root: {path!r}") | |
| return candidate | |
| def read_file( | |
| path: str, | |
| start_line: Optional[int] = None, | |
| end_line: Optional[int] = None, | |
| max_chars: int = 20000 | |
| ) -> str: | |
| """ | |
| Reads a file with optional line-based chunking. | |
| Used to prevent context overflow for large files. | |
| """ | |
| p = Path(os.path.expanduser(path)) | |
| if not p.exists(): | |
| return f"ERROR: File {path} does not exist." | |
| if not p.is_file(): | |
| return f"ERROR: {path} is not a file." | |
| ext = p.suffix.lower() | |
| try: | |
| if ext in (".txt", ".md", ".py", ".sh", ".json", ".yaml", ".yml"): | |
| lines = p.read_text(encoding="utf-8", errors="replace").splitlines() | |
| if start_line is not None or end_line is not None: | |
| start = (start_line - 1) if start_line else 0 | |
| end = end_line if end_line else len(lines) | |
| text = "\n".join(lines[start:end]) | |
| if len(text) > max_chars: | |
| text = text[:max_chars] + "\n...[truncated]" | |
| return text | |
| text = "\n".join(lines) | |
| elif ext == ".csv": | |
| df = pd.read_csv(p) | |
| text = df.to_csv(index=False) | |
| elif ext in (".xlsx", ".xls"): | |
| dfs = pd.read_excel(p, sheet_name=None) | |
| text = "\n\n".join( | |
| f"--- Sheet: {name} ---\n{df.to_csv(index=False)}" | |
| for name, df in dfs.items() | |
| ) | |
| elif ext == ".pdf": | |
| reader = PdfReader(str(p)) | |
| text = "\n".join((page.extract_text() or "") for page in reader.pages) | |
| elif ext == ".docx": | |
| doc = Document(str(p)) | |
| text = "\n".join(para.text for para in doc.paragraphs) | |
| else: | |
| try: | |
| text = p.read_text(encoding="utf-8", errors="replace") | |
| except Exception: | |
| return f"[binary file, {p.stat().st_size} bytes]" | |
| except Exception as e: | |
| return f"ERROR reading {path}: {e}" | |
| if len(text) > max_chars: | |
| text = text[:max_chars] + "\n...[truncated]" | |
| return text | |
| def ls(path: str = ".") -> str: | |
| """List contents of a directory.""" | |
| p = Path(os.path.expanduser(path)) | |
| if not p.exists(): | |
| return f"ERROR: Path {path} does not exist." | |
| if not p.is_dir(): | |
| return f"ERROR: {path} is not a directory." | |
| try: | |
| items = [] | |
| for item in p.iterdir(): | |
| suffix = "/" if item.is_dir() else "" | |
| items.append(f"{item.name}{suffix}") | |
| return "\n".join(sorted(items)) | |
| except Exception as e: | |
| return f"ERROR listing {path}: {e}" | |
| def grep(pattern: str, path: str, ignore_case: bool = True) -> str: | |
| """Search for a pattern in a file and return matching lines with line numbers.""" | |
| p = Path(os.path.expanduser(path)) | |
| if not p.exists(): | |
| return f"ERROR: File {path} does not exist." | |
| try: | |
| matches = [] | |
| lines = p.read_text(encoding="utf-8", errors="replace").splitlines() | |
| for i, line in enumerate(lines, 1): | |
| if ignore_case: | |
| if pattern.lower() in line.lower(): | |
| matches.append(f"Line {i}: {line}") | |
| else: | |
| if pattern in line: | |
| matches.append(f"Line {i}: {line}") | |
| if not matches: | |
| return "No matches found." | |
| return "\n".join(matches) | |
| except Exception as e: | |
| return f"ERROR grepping {path}: {e}" | |
| def glob_files(pattern: str, root_dir: str = ".") -> str: | |
| """Find files matching a glob pattern.""" | |
| pattern = os.path.expanduser(pattern) | |
| root_dir = os.path.expanduser(root_dir) | |
| try: | |
| if os.path.isabs(pattern): | |
| matches = _glob.glob(pattern, recursive=True) | |
| else: | |
| matches = _glob.glob(pattern, root_dir=root_dir, recursive=True) | |
| if not matches: | |
| return "No files found matching the pattern." | |
| return "\n".join(sorted(matches)) | |
| except Exception as e: | |
| return f"ERROR globbing {pattern}: {e}" | |
| def find_files(name: str, root: str = ".", max_results: int = 200) -> str: | |
| """Locate files by name under `root` using unix `find`. Fast deep search.""" | |
| root = os.path.expanduser(root) | |
| if not Path(root).exists(): | |
| return f"ERROR: Path {root} does not exist." | |
| argv = ["find", str(root), "-name", name] | |
| for excl in _DEFAULT_EXCLUDES: | |
| argv.extend(["-not", "-path", f"*/{excl}/*"]) | |
| try: | |
| result = subprocess.run( | |
| argv, | |
| capture_output=True, | |
| text=True, | |
| timeout=30, | |
| ) | |
| lines = [ln for ln in result.stdout.splitlines() if ln] | |
| if not lines: | |
| return "No files found." | |
| if len(lines) > max_results: | |
| shown = lines[:max_results] | |
| shown.append( | |
| f"[truncated: showing first {max_results} of {len(lines)} matches]" | |
| ) | |
| return "\n".join(shown) | |
| return "\n".join(lines) | |
| except subprocess.TimeoutExpired: | |
| return "ERROR: find timed out (>30s)." | |
| except Exception as e: | |
| return f"ERROR: {e}" | |
| def write_file(path: str, content: str) -> str: | |
| """Write content to a file inside the sandboxed write root. | |
| The agent may only write under the configured write root (default | |
| `.lilith/scratch`). Absolute paths and `..`-escapes are rejected. | |
| """ | |
| try: | |
| safe = _resolve_safe_write_path(path) | |
| except ValueError as exc: | |
| return f"ERROR writing to {path}: {exc}" | |
| try: | |
| safe.parent.mkdir(parents=True, exist_ok=True) | |
| safe.write_text(content, encoding="utf-8") | |
| return f"Successfully wrote {len(content)} characters to {safe}." | |
| except Exception as e: | |
| return f"ERROR writing to {path}: {e}" | |