mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-27 04:20:09 +00:00
zeekay
ea3116aac9
- Remove all unconditional pytest.skip() calls - Fix file content tests to handle both strict/non-strict modes - Fix missing device_capabilities module in hanzo-network - Fix HanzoNetProvider imports and exports - Fix local_llm.py import errors - Create test helper classes for pagination tests - Fix FastMCP paginator test constructor args - Configure all API keys from ~/.zsh/after.zsh - All tests now pass or skip conditionally based on actual requirements
120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
"""History management for file edits with disk-based storage and memory constraints."""
|
|
|
|
import logging
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from .file_cache import FileCache
|
|
|
|
|
|
class FileHistoryManager:
|
|
"""Manages file edit history with disk-based storage and memory constraints."""
|
|
|
|
def __init__(
|
|
self, max_history_per_file: int = 5, history_dir: Optional[Path] = None
|
|
):
|
|
"""Initialize the history manager.
|
|
|
|
Args:
|
|
max_history_per_file: Maximum number of history entries to keep per file (default: 5)
|
|
history_dir: Directory to store history files. If None, uses a temp directory
|
|
|
|
Notes:
|
|
- Each file's history is limited to the last N entries to conserve memory
|
|
- The file cache is limited to prevent excessive disk usage
|
|
- Older entries are automatically removed when limits are exceeded
|
|
"""
|
|
self.max_history_per_file = max_history_per_file
|
|
if history_dir is None:
|
|
history_dir = Path(tempfile.mkdtemp(prefix="oh_editor_history_"))
|
|
self.cache = FileCache(str(history_dir))
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
def _get_metadata_key(self, file_path: Path) -> str:
|
|
return f"{file_path}.metadata"
|
|
|
|
def _get_history_key(self, file_path: Path, counter: int) -> str:
|
|
return f"{file_path}.{counter}"
|
|
|
|
def add_history(self, file_path: Path, content: str):
|
|
"""Add a new history entry for a file."""
|
|
metadata_key = self._get_metadata_key(file_path)
|
|
metadata = self.cache.get(metadata_key, {"entries": [], "counter": 0})
|
|
counter = metadata["counter"]
|
|
|
|
# Add new entry
|
|
history_key = self._get_history_key(file_path, counter)
|
|
self.cache.set(history_key, content)
|
|
|
|
metadata["entries"].append(counter)
|
|
metadata["counter"] += 1
|
|
|
|
# Keep only last N entries
|
|
while len(metadata["entries"]) > self.max_history_per_file:
|
|
old_counter = metadata["entries"].pop(0)
|
|
old_history_key = self._get_history_key(file_path, old_counter)
|
|
self.cache.delete(old_history_key)
|
|
|
|
self.cache.set(metadata_key, metadata)
|
|
|
|
def pop_last_history(self, file_path: Path) -> Optional[str]:
|
|
"""Pop and return the most recent history entry for a file."""
|
|
metadata_key = self._get_metadata_key(file_path)
|
|
metadata = self.cache.get(metadata_key, {"entries": [], "counter": 0})
|
|
entries = metadata["entries"]
|
|
|
|
if not entries:
|
|
return None
|
|
|
|
# Pop and remove the last entry
|
|
last_counter = entries.pop()
|
|
history_key = self._get_history_key(file_path, last_counter)
|
|
content = self.cache.get(history_key)
|
|
|
|
if content is None:
|
|
self.logger.warning(f"History entry not found for {file_path}")
|
|
else:
|
|
# Remove the entry from the cache
|
|
self.cache.delete(history_key)
|
|
|
|
# Update metadata
|
|
metadata["entries"] = entries
|
|
self.cache.set(metadata_key, metadata)
|
|
|
|
return content
|
|
|
|
def get_metadata(self, file_path: Path):
|
|
"""Get metadata for a file (for testing purposes)."""
|
|
metadata_key = self._get_metadata_key(file_path)
|
|
metadata = self.cache.get(metadata_key, {"entries": [], "counter": 0})
|
|
return metadata # Return the actual metadata, not a copy
|
|
|
|
def clear_history(self, file_path: Path):
|
|
"""Clear history for a given file."""
|
|
metadata_key = self._get_metadata_key(file_path)
|
|
metadata = self.cache.get(metadata_key, {"entries": [], "counter": 0})
|
|
|
|
# Delete all history entries
|
|
for counter in metadata["entries"]:
|
|
history_key = self._get_history_key(file_path, counter)
|
|
self.cache.delete(history_key)
|
|
|
|
# Clear metadata
|
|
self.cache.set(metadata_key, {"entries": [], "counter": 0})
|
|
|
|
def get_all_history(self, file_path: Path) -> List[str]:
|
|
"""Get all history entries for a file."""
|
|
metadata_key = self._get_metadata_key(file_path)
|
|
metadata = self.cache.get(metadata_key, {"entries": [], "counter": 0})
|
|
entries = metadata["entries"]
|
|
|
|
history = []
|
|
for counter in entries:
|
|
history_key = self._get_history_key(file_path, counter)
|
|
content = self.cache.get(history_key)
|
|
if content is not None:
|
|
history.append(content)
|
|
|
|
return history
|