init commit
This commit is contained in:
30
app/utils/__init__.py
Normal file
30
app/utils/__init__.py
Normal file
@@ -0,0 +1,30 @@
|
||||
"""SneakyCode shared utilities."""
|
||||
|
||||
from app.utils.file_helpers import (
|
||||
BinaryFileError,
|
||||
FileSizeError,
|
||||
PathSecurityError,
|
||||
check_file_size,
|
||||
is_binary_file,
|
||||
resolve_safe_path,
|
||||
safe_read_file,
|
||||
safe_write_file,
|
||||
)
|
||||
from app.utils.logging import console, get_logger, setup_logging
|
||||
from app.utils.token_counter import TokenCounter, TokenUsage
|
||||
|
||||
__all__ = [
|
||||
"BinaryFileError",
|
||||
"FileSizeError",
|
||||
"PathSecurityError",
|
||||
"TokenCounter",
|
||||
"TokenUsage",
|
||||
"check_file_size",
|
||||
"console",
|
||||
"get_logger",
|
||||
"is_binary_file",
|
||||
"resolve_safe_path",
|
||||
"safe_read_file",
|
||||
"safe_write_file",
|
||||
"setup_logging",
|
||||
]
|
||||
103
app/utils/display.py
Normal file
103
app/utils/display.py
Normal file
@@ -0,0 +1,103 @@
|
||||
"""Rich terminal display helpers for SneakyCode."""
|
||||
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich.theme import Theme
|
||||
|
||||
from app.models.message import Message
|
||||
from app.utils.logging import console
|
||||
|
||||
# Custom theme for consistent styling across the application
|
||||
SNEAKYCODE_THEME = Theme(
|
||||
{
|
||||
"info": "cyan",
|
||||
"warning": "yellow",
|
||||
"error": "bold red",
|
||||
"success": "bold green",
|
||||
"tool": "magenta",
|
||||
"dim": "dim white",
|
||||
}
|
||||
)
|
||||
|
||||
# Apply the theme to the shared console
|
||||
console.push_theme(SNEAKYCODE_THEME)
|
||||
|
||||
|
||||
def print_banner() -> None:
|
||||
"""Print the SneakyCode startup banner."""
|
||||
console.print(
|
||||
"\n[bold cyan] SneakyCode[/bold cyan] [dim]— Local AI Coding Agent[/dim]\n",
|
||||
)
|
||||
|
||||
|
||||
def print_info(message: str) -> None:
|
||||
"""Print an informational message."""
|
||||
console.print(f"[info]{message}[/info]")
|
||||
|
||||
|
||||
def print_warning(message: str) -> None:
|
||||
"""Print a warning message."""
|
||||
console.print(f"[warning]⚠ {message}[/warning]")
|
||||
|
||||
|
||||
def print_error(message: str) -> None:
|
||||
"""Print an error message."""
|
||||
console.print(f"[error]✗ {message}[/error]")
|
||||
|
||||
|
||||
def print_success(message: str) -> None:
|
||||
"""Print a success message."""
|
||||
console.print(f"[success]✓ {message}[/success]")
|
||||
|
||||
|
||||
def print_user_message(content: str) -> None:
|
||||
"""Print a user message in a styled panel."""
|
||||
console.print(Panel(content, title="You", border_style="cyan", expand=False))
|
||||
|
||||
|
||||
def print_assistant_message(content: str) -> None:
|
||||
"""Print an assistant message in a styled panel."""
|
||||
console.print(Panel(content, title="Assistant", border_style="green", expand=False))
|
||||
|
||||
|
||||
def print_tool_call(name: str, args: str) -> None:
|
||||
"""Print a tool call summary (stub for Phase 4)."""
|
||||
console.print(f"[tool]Tool: {name}[/tool] [dim]{args}[/dim]")
|
||||
|
||||
|
||||
def print_token_usage(usage_tokens: int, budget: int) -> None:
|
||||
"""Print current token usage against budget."""
|
||||
console.print(f"[dim]Tokens: ~{usage_tokens:,} / {budget:,}[/dim]")
|
||||
|
||||
|
||||
def print_history(messages: list[Message]) -> None:
|
||||
"""Print conversation history as a Rich table.
|
||||
|
||||
Args:
|
||||
messages: List of conversation messages to display.
|
||||
"""
|
||||
if not messages:
|
||||
console.print("[dim]No messages in history.[/dim]")
|
||||
return
|
||||
|
||||
table = Table(title="Conversation History")
|
||||
table.add_column("#", style="dim", width=4)
|
||||
table.add_column("Role", width=10)
|
||||
table.add_column("Content")
|
||||
|
||||
role_styles = {
|
||||
"user": "cyan",
|
||||
"assistant": "green",
|
||||
"system": "yellow",
|
||||
"tool": "magenta",
|
||||
}
|
||||
|
||||
for i, msg in enumerate(messages, 1):
|
||||
style = role_styles.get(msg.role, "white")
|
||||
content = msg.content or "[dim](no content)[/dim]"
|
||||
# Truncate long content for display
|
||||
if len(content) > 120:
|
||||
content = content[:117] + "..."
|
||||
table.add_row(str(i), f"[{style}]{msg.role}[/{style}]", content)
|
||||
|
||||
console.print(table)
|
||||
150
app/utils/file_helpers.py
Normal file
150
app/utils/file_helpers.py
Normal file
@@ -0,0 +1,150 @@
|
||||
"""Security-critical file operation helpers with path sandboxing."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class PathSecurityError(Exception):
|
||||
"""Raised when a file path escapes the allowed workspace root."""
|
||||
|
||||
|
||||
class FileSizeError(Exception):
|
||||
"""Raised when a file exceeds the configured size limit."""
|
||||
|
||||
|
||||
class BinaryFileError(Exception):
|
||||
"""Raised when an operation is attempted on a binary file."""
|
||||
|
||||
|
||||
def resolve_safe_path(path: str | Path, workspace_root: Path) -> Path:
|
||||
"""Resolve a path and verify it is within the workspace root.
|
||||
|
||||
Args:
|
||||
path: The path to resolve (absolute or relative to workspace_root).
|
||||
workspace_root: The allowed root directory (must be absolute).
|
||||
|
||||
Returns:
|
||||
The resolved absolute path.
|
||||
|
||||
Raises:
|
||||
PathSecurityError: If the resolved path is outside workspace_root.
|
||||
"""
|
||||
workspace_root = workspace_root.resolve()
|
||||
resolved = (workspace_root / path).resolve()
|
||||
|
||||
if not resolved.is_relative_to(workspace_root):
|
||||
raise PathSecurityError(
|
||||
f"Path '{path}' resolves to '{resolved}' which is outside "
|
||||
f"workspace root '{workspace_root}'"
|
||||
)
|
||||
|
||||
return resolved
|
||||
|
||||
|
||||
def is_binary_file(file_path: Path, sample_size: int = 8192) -> bool:
|
||||
"""Detect if a file is binary by checking for null bytes in a sample.
|
||||
|
||||
Args:
|
||||
file_path: Path to the file to check.
|
||||
sample_size: Number of bytes to read for detection.
|
||||
|
||||
Returns:
|
||||
True if the file appears to be binary.
|
||||
"""
|
||||
try:
|
||||
with open(file_path, "rb") as f:
|
||||
sample = f.read(sample_size)
|
||||
return b"\x00" in sample
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def check_file_size(file_path: Path, max_size_bytes: int) -> None:
|
||||
"""Verify that a file does not exceed the size limit.
|
||||
|
||||
Args:
|
||||
file_path: Path to the file to check.
|
||||
max_size_bytes: Maximum allowed file size in bytes.
|
||||
|
||||
Raises:
|
||||
FileSizeError: If the file exceeds the size limit.
|
||||
FileNotFoundError: If the file does not exist.
|
||||
"""
|
||||
size = file_path.stat().st_size
|
||||
if size > max_size_bytes:
|
||||
raise FileSizeError(
|
||||
f"File '{file_path}' is {size:,} bytes, exceeding the "
|
||||
f"{max_size_bytes:,} byte limit"
|
||||
)
|
||||
|
||||
|
||||
def safe_read_file(
|
||||
path: str | Path,
|
||||
workspace_root: Path,
|
||||
max_size_bytes: int = 1_048_576,
|
||||
check_binary: bool = True,
|
||||
) -> str:
|
||||
"""Safely read a file with path sandboxing, size, and binary checks.
|
||||
|
||||
Args:
|
||||
path: Path to the file (relative to workspace_root or absolute).
|
||||
workspace_root: The allowed root directory.
|
||||
max_size_bytes: Maximum file size to read.
|
||||
check_binary: Whether to reject binary files.
|
||||
|
||||
Returns:
|
||||
The file contents as a string.
|
||||
|
||||
Raises:
|
||||
PathSecurityError: If the path escapes the workspace.
|
||||
FileSizeError: If the file is too large.
|
||||
BinaryFileError: If the file is binary and check_binary is True.
|
||||
FileNotFoundError: If the file does not exist.
|
||||
"""
|
||||
safe_path = resolve_safe_path(path, workspace_root)
|
||||
|
||||
if not safe_path.exists():
|
||||
raise FileNotFoundError(f"File not found: {safe_path}")
|
||||
|
||||
check_file_size(safe_path, max_size_bytes)
|
||||
|
||||
if check_binary and is_binary_file(safe_path):
|
||||
raise BinaryFileError(f"File appears to be binary: {safe_path}")
|
||||
|
||||
return safe_path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def safe_write_file(
|
||||
path: str | Path,
|
||||
content: str,
|
||||
workspace_root: Path,
|
||||
max_size_bytes: int = 1_048_576,
|
||||
) -> Path:
|
||||
"""Safely write a file with path sandboxing and size checks.
|
||||
|
||||
Args:
|
||||
path: Path to write to (relative to workspace_root or absolute).
|
||||
content: String content to write.
|
||||
workspace_root: The allowed root directory.
|
||||
max_size_bytes: Maximum allowed content size in bytes.
|
||||
|
||||
Returns:
|
||||
The resolved path that was written to.
|
||||
|
||||
Raises:
|
||||
PathSecurityError: If the path escapes the workspace.
|
||||
FileSizeError: If the content exceeds the size limit.
|
||||
"""
|
||||
safe_path = resolve_safe_path(path, workspace_root)
|
||||
|
||||
content_size = len(content.encode("utf-8"))
|
||||
if content_size > max_size_bytes:
|
||||
raise FileSizeError(
|
||||
f"Content is {content_size:,} bytes, exceeding the "
|
||||
f"{max_size_bytes:,} byte limit"
|
||||
)
|
||||
|
||||
# Ensure parent directory exists
|
||||
safe_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
safe_path.write_text(content, encoding="utf-8")
|
||||
|
||||
return safe_path
|
||||
89
app/utils/logging.py
Normal file
89
app/utils/logging.py
Normal file
@@ -0,0 +1,89 @@
|
||||
"""Centralized logging setup: structlog for file logs, Rich for terminal output."""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
import structlog
|
||||
from rich.console import Console
|
||||
|
||||
# Shared Rich console instance for the entire application
|
||||
console = Console()
|
||||
|
||||
|
||||
def setup_logging(
|
||||
log_file: Path | None = None,
|
||||
log_level: str = "INFO",
|
||||
verbose: bool = False,
|
||||
) -> None:
|
||||
"""Configure dual logging: Rich handler for terminal, optional file handler.
|
||||
|
||||
Args:
|
||||
log_file: Optional path to a log file for structured file logging.
|
||||
log_level: Minimum log level (DEBUG, INFO, WARNING, ERROR).
|
||||
verbose: If True, sets log level to DEBUG regardless of log_level.
|
||||
"""
|
||||
effective_level = "DEBUG" if verbose else log_level.upper()
|
||||
console_level = effective_level if verbose else "WARNING"
|
||||
|
||||
# Configure stdlib root logger
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(effective_level)
|
||||
|
||||
# Clear any existing handlers to avoid duplicates on re-init
|
||||
root_logger.handlers.clear()
|
||||
|
||||
# Rich handler for terminal output — only warnings+ unless verbose
|
||||
from rich.logging import RichHandler
|
||||
|
||||
rich_handler = RichHandler(
|
||||
console=console,
|
||||
show_time=True,
|
||||
show_path=verbose,
|
||||
markup=True,
|
||||
rich_tracebacks=True,
|
||||
)
|
||||
rich_handler.setLevel(console_level)
|
||||
root_logger.addHandler(rich_handler)
|
||||
|
||||
# Optional file handler for persistent structured logs
|
||||
if log_file is not None:
|
||||
log_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
file_handler = logging.FileHandler(str(log_file), encoding="utf-8")
|
||||
file_handler.setLevel(effective_level)
|
||||
file_formatter = logging.Formatter(
|
||||
"%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S",
|
||||
)
|
||||
file_handler.setFormatter(file_formatter)
|
||||
root_logger.addHandler(file_handler)
|
||||
|
||||
# Configure structlog to route through stdlib logging
|
||||
structlog.configure(
|
||||
processors=[
|
||||
structlog.contextvars.merge_contextvars,
|
||||
structlog.stdlib.filter_by_level,
|
||||
structlog.stdlib.add_logger_name,
|
||||
structlog.stdlib.add_log_level,
|
||||
structlog.stdlib.PositionalArgumentsFormatter(),
|
||||
structlog.processors.TimeStamper(fmt="iso"),
|
||||
structlog.processors.StackInfoRenderer(),
|
||||
structlog.processors.format_exc_info,
|
||||
structlog.processors.UnicodeDecoder(),
|
||||
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
|
||||
],
|
||||
logger_factory=structlog.stdlib.LoggerFactory(),
|
||||
wrapper_class=structlog.stdlib.BoundLogger,
|
||||
cache_logger_on_first_use=True,
|
||||
)
|
||||
|
||||
|
||||
def get_logger(name: str) -> structlog.stdlib.BoundLogger:
|
||||
"""Get a named structlog logger.
|
||||
|
||||
Args:
|
||||
name: Logger name, typically __name__ of the calling module.
|
||||
|
||||
Returns:
|
||||
A bound structlog logger instance.
|
||||
"""
|
||||
return structlog.get_logger(name)
|
||||
93
app/utils/token_counter.py
Normal file
93
app/utils/token_counter.py
Normal file
@@ -0,0 +1,93 @@
|
||||
"""Approximate token counting for conversation budget management."""
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.models.message import Message
|
||||
|
||||
|
||||
class TokenUsage(BaseModel):
|
||||
"""Snapshot of token usage for a single LLM call."""
|
||||
|
||||
prompt_tokens: int = Field(default=0, description="Tokens in the prompt")
|
||||
completion_tokens: int = Field(default=0, description="Tokens in the completion")
|
||||
total_tokens: int = Field(default=0, description="Total tokens used")
|
||||
|
||||
|
||||
class TokenCounter:
|
||||
"""Tracks cumulative token usage with character-based estimation.
|
||||
|
||||
Uses a simple heuristic of ~4 characters per token for estimation.
|
||||
This is intentionally approximate — accurate enough for budget tracking.
|
||||
"""
|
||||
|
||||
CHARS_PER_TOKEN: int = 4
|
||||
|
||||
def __init__(self, budget: int = 32_000) -> None:
|
||||
"""Initialize the token counter.
|
||||
|
||||
Args:
|
||||
budget: Maximum token budget for the conversation.
|
||||
"""
|
||||
self._budget = budget
|
||||
self._cumulative = TokenUsage()
|
||||
|
||||
@property
|
||||
def budget(self) -> int:
|
||||
"""The configured token budget."""
|
||||
return self._budget
|
||||
|
||||
@property
|
||||
def cumulative_usage(self) -> TokenUsage:
|
||||
"""Cumulative token usage across all tracked calls."""
|
||||
return self._cumulative
|
||||
|
||||
@property
|
||||
def remaining_budget(self) -> int:
|
||||
"""Estimated tokens remaining before hitting the budget."""
|
||||
return max(0, self._budget - self._cumulative.total_tokens)
|
||||
|
||||
def estimate_tokens(self, text: str) -> int:
|
||||
"""Estimate token count for a string using character heuristic.
|
||||
|
||||
Args:
|
||||
text: The text to estimate tokens for.
|
||||
|
||||
Returns:
|
||||
Estimated token count.
|
||||
"""
|
||||
return max(1, len(text) // self.CHARS_PER_TOKEN)
|
||||
|
||||
def estimate_messages_tokens(self, messages: list[Message]) -> int:
|
||||
"""Estimate total tokens for a list of messages.
|
||||
|
||||
Args:
|
||||
messages: List of conversation messages.
|
||||
|
||||
Returns:
|
||||
Estimated total token count.
|
||||
"""
|
||||
total = 0
|
||||
for msg in messages:
|
||||
if msg.content:
|
||||
total += self.estimate_tokens(msg.content)
|
||||
if msg.tool_calls:
|
||||
for tc in msg.tool_calls:
|
||||
total += self.estimate_tokens(tc.function.name)
|
||||
total += self.estimate_tokens(tc.function.arguments)
|
||||
# Per-message overhead (role, formatting)
|
||||
total += 4
|
||||
return total
|
||||
|
||||
def count_usage(self, usage: TokenUsage) -> None:
|
||||
"""Record token usage from an LLM call.
|
||||
|
||||
Args:
|
||||
usage: Token usage from a single call.
|
||||
"""
|
||||
self._cumulative.prompt_tokens += usage.prompt_tokens
|
||||
self._cumulative.completion_tokens += usage.completion_tokens
|
||||
self._cumulative.total_tokens += usage.total_tokens
|
||||
|
||||
def is_over_budget(self) -> bool:
|
||||
"""Check if cumulative usage has exceeded the token budget."""
|
||||
return self._cumulative.total_tokens >= self._budget
|
||||
Reference in New Issue
Block a user