"""Resolve @path references inside user input.
Supported forms:
@path/to/file.py → file content wrapped in code fence
@dir/ → list of files in directory (recursive if trailing /)
@tests/**/*.py → glob expansion, files only
Size limits apply per-file and in total to avoid flooding the LLM context.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable
MAX_FILE_BYTES = 64_000
MAX_TOTAL_BYTES = 128_000
TRUNCATED_NOTICE = "\n... [truncated by Navi Code]"
@dataclass
class ResolvedFile:
"""A file resolved from an @ reference."""
path: Path
display_path: str
content: str
truncated: bool = False
@dataclass
class FileRefResult:
"""Result of resolving @ references in a prompt."""
prompt: str # user-visible prompt (with @ markers replaced by file list)
attachments: list[ResolvedFile] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
total_bytes: int = 0
def is_empty(self) -> bool:
return not self.attachments and not self.errors
def to_message(self) -> str:
"""Build the full message to send to the backend."""
if not self.attachments and not self.errors:
return self.prompt
parts = [self.prompt]
if self.attachments:
parts.append("")
parts.append("--- attached files ---")
for f in self.attachments:
lang = _guess_language(f.path)
label = f"file: {f.display_path}"
if f.truncated:
label += " (truncated)"
parts.append(f"```{lang} {label}")
parts.append(f.content)
parts.append("```")
if self.errors:
parts.append("")
parts.append("--- attachment errors ---")
for err in self.errors:
parts.append(f"- {err}")
return "\n".join(parts)
_ref_pattern = re.compile(r"@((?:[A-Za-z0-9_\-\.~/$*?]|\\\s)+)")
def find_refs(text: str) -> list[str]:
"""Return all @path tokens found in text, in order, without duplicates."""
seen: set[str] = set()
refs: list[str] = OrderedRefs()
for raw in _ref_pattern.findall(text):
# Un-escape backslash-space inside the token.
ref = raw.replace("\\ ", " ")
if ref not in seen:
seen.add(ref)
refs.append(ref)
return refs
class OrderedRefs(list):
"""Stub kept for type clarity; plain list suffices."""
class FileRefResolver:
"""Resolve @ references relative to a base directory."""
def __init__(self, base_dir: Path | str | None = None) -> None:
self.base_dir = Path(base_dir or Path.cwd()).expanduser().resolve()
def resolve(self, text: str) -> FileRefResult:
refs = find_refs(text)
if not refs:
return FileRefResult(prompt=text)
result = FileRefResult(prompt=text)
for ref in refs:
self._resolve_ref(ref, result)
if result.total_bytes >= MAX_TOTAL_BYTES:
result.errors.append("total attachment size limit reached; remaining files skipped")
break
return result
def _resolve_ref(self, ref: str, result: FileRefResult) -> None:
path = self._expand_path(ref)
if path is None:
result.errors.append(f"could not resolve {ref!r}")
return
if path.exists():
if path.is_dir():
files = sorted(_collect_files(path, recursive=ref.endswith("/")))
if not files:
result.errors.append(f"no files found in {ref}")
return
for file_path in files:
self._attach_file(file_path, result, root_dir=path)
if result.total_bytes >= MAX_TOTAL_BYTES:
return
return
if path.is_file():
self._attach_file(path, result)
return
result.errors.append(f"not a file or directory: {ref}")
return
# Non-existent path: try glob expansion if it looks like a pattern.
if _is_glob(ref):
matches = sorted(self.base_dir.glob(ref))
if not matches:
result.errors.append(f"no matches for {ref}")
return
for file_path in matches:
if not file_path.is_file():
continue
self._attach_file(file_path, result)
if result.total_bytes >= MAX_TOTAL_BYTES:
return
return
result.errors.append(f"not found: {ref}")
def _attach_file(self, path: Path, result: FileRefResult, root_dir: Path | None = None) -> None:
display = _relative_or_absolute(path, self.base_dir)
if root_dir is not None:
display = _relative_or_absolute(path, root_dir)
try:
data = path.read_bytes()
except Exception as exc:
result.errors.append(f"failed to read {path}: {exc}")
return
truncated = False
if len(data) > MAX_FILE_BYTES:
data = data[:MAX_FILE_BYTES]
truncated = True
remaining = MAX_TOTAL_BYTES - result.total_bytes
if remaining <= 0:
return
if len(data) > remaining:
data = data[:remaining]
truncated = True
try:
text = data.decode("utf-8", errors="replace")
except Exception as exc:
result.errors.append(f"failed to decode {path}: {exc}")
return
if truncated:
text += TRUNCATED_NOTICE
result.attachments.append(ResolvedFile(path=path, display_path=display, content=text, truncated=truncated))
result.total_bytes += len(data)
def _expand_path(self, ref: str) -> Path | None:
# Strip any trailing slash for expansion, but keep the flag later.
clean = ref.rstrip("/")
if not clean:
return None
if clean.startswith("~"):
return Path(clean).expanduser()
return (self.base_dir / clean).resolve()
def _is_glob(ref: str) -> bool:
"""Return True if ref contains glob metacharacters."""
return "*" in ref or "?" in ref or "[" in ref
def _collect_files(path: Path, recursive: bool = False) -> Iterable[Path]:
"""Yield files inside a directory."""
if recursive:
for p in sorted(path.rglob("*")):
if p.is_file():
yield p
else:
for p in sorted(path.iterdir()):
if p.is_file():
yield p
def _relative_or_absolute(path: Path, base: Path) -> str:
try:
return str(path.relative_to(base))
except ValueError:
return str(path)
def _guess_language(path: Path) -> str:
"""Best-effort language tag for markdown code fence."""
mapping = {
".py": "python",
".js": "javascript",
".ts": "typescript",
".tsx": "tsx",
".jsx": "jsx",
".go": "go",
".rs": "rust",
".c": "c",
".cpp": "cpp",
".h": "c",
".java": "java",
".kt": "kotlin",
".sh": "bash",
".zsh": "bash",
".bash": "bash",
".md": "markdown",
".json": "json",
".yaml": "yaml",
".yml": "yaml",
".toml": "toml",
".html": "html",
".css": "css",
".scss": "scss",
".sql": "sql",
".dockerfile": "dockerfile",
".lock": "text",
".txt": "text",
".env": "bash",
}
return mapping.get(path.suffix.lower(), "text")
