Initial commit: Claude Code Hooks with Diátaxis documentation
✨ Features: - 🧠 Shadow learner that builds intelligence from command patterns - 🛡️ Smart command validation with safety checks - 💾 Automatic context monitoring and backup system - 🔄 Session continuity across Claude restarts 📚 Documentation: - Complete Diátaxis-organized documentation - Learning-oriented tutorial for getting started - Task-oriented how-to guides for specific problems - Information-oriented reference for quick lookup - Understanding-oriented explanations of architecture 🚀 Installation: - One-command installation script - Bootstrap prompt for installation via Claude - Cross-platform compatibility - Comprehensive testing suite 🎯 Ready for real-world use and community feedback! 🤖 Generated with Claude Code Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
commit
162ca67098
34 changed files with 5904 additions and 0 deletions
21
lib/__init__.py
Normal file
21
lib/__init__.py
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
"""Claude Code Hooks System - Core Library"""
|
||||
|
||||
__version__ = "1.0.0"
|
||||
__author__ = "Claude Code Hooks Contributors"
|
||||
|
||||
from .models import *
|
||||
from .shadow_learner import ShadowLearner
|
||||
from .context_monitor import ContextMonitor
|
||||
from .backup_manager import BackupManager
|
||||
from .session_state import SessionStateManager
|
||||
|
||||
__all__ = [
|
||||
"ShadowLearner",
|
||||
"ContextMonitor",
|
||||
"BackupManager",
|
||||
"SessionStateManager",
|
||||
"Pattern",
|
||||
"ToolExecution",
|
||||
"HookResult",
|
||||
"ValidationResult"
|
||||
]
|
||||
388
lib/backup_manager.py
Normal file
388
lib/backup_manager.py
Normal file
|
|
@ -0,0 +1,388 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Backup Manager - Resilient backup execution system"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, List
|
||||
|
||||
try:
|
||||
from .models import BackupResult, GitBackupResult, BackupDecision
|
||||
except ImportError:
|
||||
from models import BackupResult, GitBackupResult, BackupDecision
|
||||
|
||||
|
||||
class BackupManager:
|
||||
"""Handles backup execution with comprehensive error handling"""
|
||||
|
||||
def __init__(self, project_root: str = "."):
|
||||
self.project_root = Path(project_root).resolve()
|
||||
self.backup_dir = self.project_root / ".claude_hooks" / "backups"
|
||||
self.backup_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Backup settings
|
||||
self.max_backups = 10
|
||||
self.log_file = self.backup_dir / "backup.log"
|
||||
|
||||
def execute_backup(self, decision: BackupDecision,
|
||||
session_state: Dict[str, Any]) -> BackupResult:
|
||||
"""Execute backup with comprehensive error handling"""
|
||||
|
||||
backup_id = self._generate_backup_id()
|
||||
backup_path = self.backup_dir / backup_id
|
||||
|
||||
try:
|
||||
# Create backup structure
|
||||
backup_info = self._create_backup_structure(backup_path, session_state)
|
||||
|
||||
# Git backup (if possible)
|
||||
git_result = self._attempt_git_backup(backup_id, decision.reason)
|
||||
|
||||
# File system backup
|
||||
fs_result = self._create_filesystem_backup(backup_path, session_state)
|
||||
|
||||
# Session state backup
|
||||
state_result = self._backup_session_state(backup_path, session_state)
|
||||
|
||||
# Clean up old backups
|
||||
self._cleanup_old_backups()
|
||||
|
||||
# Log successful backup
|
||||
self._log_backup(backup_id, decision, success=True)
|
||||
|
||||
return BackupResult(
|
||||
success=True,
|
||||
backup_id=backup_id,
|
||||
backup_path=str(backup_path),
|
||||
git_success=git_result.success,
|
||||
components={
|
||||
"git": git_result,
|
||||
"filesystem": fs_result,
|
||||
"session_state": state_result
|
||||
}
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
# Backup failures should never break the session
|
||||
fallback_result = self._create_minimal_backup(session_state)
|
||||
self._log_backup(backup_id, decision, success=False, error=str(e))
|
||||
|
||||
return BackupResult(
|
||||
success=False,
|
||||
backup_id=backup_id,
|
||||
error=str(e),
|
||||
fallback_performed=fallback_result
|
||||
)
|
||||
|
||||
def _generate_backup_id(self) -> str:
|
||||
"""Generate unique backup identifier"""
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
return f"backup_{timestamp}"
|
||||
|
||||
def _create_backup_structure(self, backup_path: Path, session_state: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Create basic backup directory structure"""
|
||||
backup_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create subdirectories
|
||||
(backup_path / "files").mkdir(exist_ok=True)
|
||||
(backup_path / "logs").mkdir(exist_ok=True)
|
||||
(backup_path / "state").mkdir(exist_ok=True)
|
||||
|
||||
# Create backup metadata
|
||||
metadata = {
|
||||
"backup_id": backup_path.name,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"session_state": session_state,
|
||||
"project_root": str(self.project_root)
|
||||
}
|
||||
|
||||
with open(backup_path / "metadata.json", 'w') as f:
|
||||
json.dump(metadata, f, indent=2)
|
||||
|
||||
return metadata
|
||||
|
||||
def _attempt_git_backup(self, backup_id: str, reason: str) -> GitBackupResult:
|
||||
"""Attempt git backup with proper error handling"""
|
||||
try:
|
||||
# Check if git repo exists
|
||||
if not (self.project_root / ".git").exists():
|
||||
# Initialize repo if none exists
|
||||
result = subprocess.run(
|
||||
["git", "init"],
|
||||
cwd=self.project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return GitBackupResult(
|
||||
success=False,
|
||||
error=f"Git init failed: {result.stderr}"
|
||||
)
|
||||
|
||||
# Add all changes
|
||||
result = subprocess.run(
|
||||
["git", "add", "-A"],
|
||||
cwd=self.project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return GitBackupResult(
|
||||
success=False,
|
||||
error=f"Git add failed: {result.stderr}"
|
||||
)
|
||||
|
||||
# Check if there are changes to commit
|
||||
result = subprocess.run(
|
||||
["git", "status", "--porcelain"],
|
||||
cwd=self.project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
if not result.stdout.strip():
|
||||
return GitBackupResult(
|
||||
success=True,
|
||||
message="No changes to commit"
|
||||
)
|
||||
|
||||
# Create commit
|
||||
commit_msg = f"Claude hooks auto-backup: {reason} ({backup_id})"
|
||||
result = subprocess.run(
|
||||
["git", "commit", "-m", commit_msg],
|
||||
cwd=self.project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
return GitBackupResult(
|
||||
success=False,
|
||||
error=f"Git commit failed: {result.stderr}"
|
||||
)
|
||||
|
||||
# Get commit ID
|
||||
commit_id = self._get_latest_commit()
|
||||
|
||||
return GitBackupResult(
|
||||
success=True,
|
||||
commit_id=commit_id,
|
||||
message=f"Committed as {commit_id[:8]}"
|
||||
)
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return GitBackupResult(
|
||||
success=False,
|
||||
error="Git operation timed out"
|
||||
)
|
||||
except subprocess.CalledProcessError as e:
|
||||
return GitBackupResult(
|
||||
success=False,
|
||||
error=f"Git error: {e}"
|
||||
)
|
||||
except Exception as e:
|
||||
return GitBackupResult(
|
||||
success=False,
|
||||
error=f"Unexpected git error: {e}"
|
||||
)
|
||||
|
||||
def _get_latest_commit(self) -> str:
|
||||
"""Get the latest commit ID"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "rev-parse", "HEAD"],
|
||||
cwd=self.project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10
|
||||
)
|
||||
if result.returncode == 0:
|
||||
return result.stdout.strip()
|
||||
except Exception:
|
||||
pass
|
||||
return "unknown"
|
||||
|
||||
def _create_filesystem_backup(self, backup_path: Path,
|
||||
session_state: Dict[str, Any]) -> BackupResult:
|
||||
"""Create filesystem backup of important files"""
|
||||
try:
|
||||
files_dir = backup_path / "files"
|
||||
files_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Backup modified files mentioned in session
|
||||
modified_files = session_state.get("modified_files", [])
|
||||
files_backed_up = []
|
||||
|
||||
for file_path in modified_files:
|
||||
try:
|
||||
src = Path(file_path)
|
||||
if src.exists() and src.is_file():
|
||||
# Create relative path structure
|
||||
rel_path = src.relative_to(self.project_root) if src.is_relative_to(self.project_root) else src.name
|
||||
dst = files_dir / rel_path
|
||||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
shutil.copy2(src, dst)
|
||||
files_backed_up.append(str(src))
|
||||
except Exception as e:
|
||||
# Log error but continue with other files
|
||||
self._log_file_backup_error(file_path, e)
|
||||
|
||||
# Backup important project files
|
||||
important_files = [
|
||||
"package.json", "requirements.txt", "Cargo.toml",
|
||||
"pyproject.toml", "setup.py", ".gitignore",
|
||||
"README.md", "CLAUDE.md"
|
||||
]
|
||||
|
||||
for file_name in important_files:
|
||||
file_path = self.project_root / file_name
|
||||
if file_path.exists():
|
||||
try:
|
||||
dst = files_dir / file_name
|
||||
shutil.copy2(file_path, dst)
|
||||
files_backed_up.append(str(file_path))
|
||||
except Exception:
|
||||
pass # Not critical
|
||||
|
||||
return BackupResult(
|
||||
success=True,
|
||||
message=f"Backed up {len(files_backed_up)} files",
|
||||
metadata={"files": files_backed_up}
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return BackupResult(success=False, error=str(e))
|
||||
|
||||
def _backup_session_state(self, backup_path: Path,
|
||||
session_state: Dict[str, Any]) -> BackupResult:
|
||||
"""Backup session state and context"""
|
||||
try:
|
||||
state_dir = backup_path / "state"
|
||||
|
||||
# Save session state
|
||||
with open(state_dir / "session.json", 'w') as f:
|
||||
json.dump(session_state, f, indent=2)
|
||||
|
||||
# Copy hook logs if they exist
|
||||
logs_source = self.project_root / ".claude_hooks" / "logs"
|
||||
if logs_source.exists():
|
||||
logs_dest = backup_path / "logs"
|
||||
shutil.copytree(logs_source, logs_dest, exist_ok=True)
|
||||
|
||||
# Copy patterns database
|
||||
patterns_source = self.project_root / ".claude_hooks" / "patterns"
|
||||
if patterns_source.exists():
|
||||
patterns_dest = state_dir / "patterns"
|
||||
shutil.copytree(patterns_source, patterns_dest, exist_ok=True)
|
||||
|
||||
return BackupResult(
|
||||
success=True,
|
||||
message="Session state backed up"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return BackupResult(success=False, error=str(e))
|
||||
|
||||
def _create_minimal_backup(self, session_state: Dict[str, Any]) -> bool:
|
||||
"""Create minimal backup when full backup fails"""
|
||||
try:
|
||||
# At minimum, save session state to a simple file
|
||||
emergency_file = self.backup_dir / "emergency_backup.json"
|
||||
|
||||
emergency_data = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"session_state": session_state,
|
||||
"type": "emergency_backup"
|
||||
}
|
||||
|
||||
with open(emergency_file, 'w') as f:
|
||||
json.dump(emergency_data, f, indent=2)
|
||||
|
||||
return True
|
||||
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _cleanup_old_backups(self):
|
||||
"""Remove old backups to save space"""
|
||||
try:
|
||||
# Get all backup directories
|
||||
backup_dirs = [d for d in self.backup_dir.iterdir()
|
||||
if d.is_dir() and d.name.startswith("backup_")]
|
||||
|
||||
# Sort by creation time (newest first)
|
||||
backup_dirs.sort(key=lambda d: d.stat().st_mtime, reverse=True)
|
||||
|
||||
# Remove old backups beyond max_backups
|
||||
for old_backup in backup_dirs[self.max_backups:]:
|
||||
shutil.rmtree(old_backup)
|
||||
|
||||
except Exception:
|
||||
pass # Cleanup failures shouldn't break backup
|
||||
|
||||
def _log_backup(self, backup_id: str, decision: BackupDecision,
|
||||
success: bool, error: str = ""):
|
||||
"""Log backup operation"""
|
||||
try:
|
||||
log_entry = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"backup_id": backup_id,
|
||||
"reason": decision.reason,
|
||||
"urgency": decision.urgency,
|
||||
"success": success,
|
||||
"error": error
|
||||
}
|
||||
|
||||
# Append to log file
|
||||
with open(self.log_file, 'a') as f:
|
||||
f.write(json.dumps(log_entry) + "\n")
|
||||
|
||||
except Exception:
|
||||
pass # Logging failures shouldn't break backup
|
||||
|
||||
def _log_file_backup_error(self, file_path: str, error: Exception):
|
||||
"""Log file backup errors"""
|
||||
try:
|
||||
error_entry = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"type": "file_backup_error",
|
||||
"file_path": file_path,
|
||||
"error": str(error)
|
||||
}
|
||||
|
||||
with open(self.log_file, 'a') as f:
|
||||
f.write(json.dumps(error_entry) + "\n")
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def list_backups(self) -> List[Dict[str, Any]]:
|
||||
"""List available backups"""
|
||||
backups = []
|
||||
|
||||
try:
|
||||
backup_dirs = [d for d in self.backup_dir.iterdir()
|
||||
if d.is_dir() and d.name.startswith("backup_")]
|
||||
|
||||
for backup_dir in backup_dirs:
|
||||
metadata_file = backup_dir / "metadata.json"
|
||||
if metadata_file.exists():
|
||||
try:
|
||||
with open(metadata_file, 'r') as f:
|
||||
metadata = json.load(f)
|
||||
backups.append(metadata)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return sorted(backups, key=lambda b: b.get("timestamp", ""), reverse=True)
|
||||
202
lib/cli.py
Normal file
202
lib/cli.py
Normal file
|
|
@ -0,0 +1,202 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Claude Hooks CLI - Command line interface for managing hooks
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from .backup_manager import BackupManager
|
||||
from .session_state import SessionStateManager
|
||||
from .shadow_learner import ShadowLearner
|
||||
from .context_monitor import ContextMonitor
|
||||
|
||||
|
||||
def list_backups():
|
||||
"""List available backups"""
|
||||
backup_manager = BackupManager()
|
||||
backups = backup_manager.list_backups()
|
||||
|
||||
if not backups:
|
||||
print("No backups found.")
|
||||
return
|
||||
|
||||
print("Available Backups:")
|
||||
print("==================")
|
||||
|
||||
for backup in backups:
|
||||
timestamp = backup.get("timestamp", "unknown")
|
||||
backup_id = backup.get("backup_id", "unknown")
|
||||
reason = backup.get("session_state", {}).get("backup_history", [])
|
||||
if reason:
|
||||
reason = reason[-1].get("reason", "unknown")
|
||||
else:
|
||||
reason = "unknown"
|
||||
|
||||
print(f"🗂️ {backup_id}")
|
||||
print(f" 📅 {timestamp}")
|
||||
print(f" 📝 {reason}")
|
||||
print()
|
||||
|
||||
|
||||
def show_session_status():
|
||||
"""Show current session status"""
|
||||
session_manager = SessionStateManager()
|
||||
context_monitor = ContextMonitor()
|
||||
|
||||
summary = session_manager.get_session_summary()
|
||||
context_summary = context_monitor.get_session_summary()
|
||||
|
||||
print("Session Status:")
|
||||
print("===============")
|
||||
print(f"Session ID: {summary.get('session_id', 'unknown')}")
|
||||
print(f"Duration: {summary.get('session_stats', {}).get('duration_minutes', 0)} minutes")
|
||||
print(f"Context Usage: {context_summary.get('context_usage_ratio', 0):.1%}")
|
||||
print(f"Tool Calls: {summary.get('session_stats', {}).get('total_tool_calls', 0)}")
|
||||
print(f"Files Modified: {len(summary.get('modified_files', []))}")
|
||||
print(f"Commands Executed: {summary.get('session_stats', {}).get('total_commands', 0)}")
|
||||
print(f"Backups Created: {len(summary.get('backup_history', []))}")
|
||||
print()
|
||||
|
||||
if summary.get('modified_files'):
|
||||
print("Modified Files:")
|
||||
for file_path in summary['modified_files']:
|
||||
print(f" - {file_path}")
|
||||
print()
|
||||
|
||||
if context_summary.get('should_backup'):
|
||||
print("⚠️ Backup recommended (high context usage)")
|
||||
else:
|
||||
print("✅ No backup needed currently")
|
||||
|
||||
|
||||
def show_patterns():
|
||||
"""Show learned patterns"""
|
||||
shadow_learner = ShadowLearner()
|
||||
|
||||
print("Learned Patterns:")
|
||||
print("=================")
|
||||
|
||||
# Command patterns
|
||||
command_patterns = shadow_learner.db.command_patterns
|
||||
if command_patterns:
|
||||
print("\n🖥️ Command Patterns:")
|
||||
for pattern_id, pattern in list(command_patterns.items())[:10]: # Show top 10
|
||||
cmd = pattern.trigger.get("command", "unknown")
|
||||
confidence = pattern.confidence
|
||||
evidence = pattern.evidence_count
|
||||
success_rate = pattern.success_rate
|
||||
|
||||
print(f" {cmd}")
|
||||
print(f" Confidence: {confidence:.1%}")
|
||||
print(f" Evidence: {evidence} samples")
|
||||
print(f" Success Rate: {success_rate:.1%}")
|
||||
|
||||
# Context patterns
|
||||
context_patterns = shadow_learner.db.context_patterns
|
||||
if context_patterns:
|
||||
print("\n🔍 Context Patterns:")
|
||||
for pattern_id, pattern in list(context_patterns.items())[:5]: # Show top 5
|
||||
error_type = pattern.trigger.get("error_type", "unknown")
|
||||
confidence = pattern.confidence
|
||||
evidence = pattern.evidence_count
|
||||
|
||||
print(f" {error_type}")
|
||||
print(f" Confidence: {confidence:.1%}")
|
||||
print(f" Evidence: {evidence} samples")
|
||||
|
||||
if not command_patterns and not context_patterns:
|
||||
print("No patterns learned yet. Use Claude Code to start building the knowledge base!")
|
||||
|
||||
|
||||
def clear_patterns():
|
||||
"""Clear learned patterns"""
|
||||
response = input("Are you sure you want to clear all learned patterns? (y/N): ")
|
||||
if response.lower() == 'y':
|
||||
shadow_learner = ShadowLearner()
|
||||
shadow_learner.db = shadow_learner._load_database() # Reset to empty
|
||||
shadow_learner.save_database()
|
||||
print("✅ Patterns cleared successfully")
|
||||
else:
|
||||
print("Operation cancelled")
|
||||
|
||||
|
||||
def export_data():
|
||||
"""Export all hook data"""
|
||||
export_dir = Path("claude_hooks_export")
|
||||
export_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Export session state
|
||||
session_manager = SessionStateManager()
|
||||
summary = session_manager.get_session_summary()
|
||||
|
||||
with open(export_dir / "session_data.json", 'w') as f:
|
||||
json.dump(summary, f, indent=2)
|
||||
|
||||
# Export patterns
|
||||
shadow_learner = ShadowLearner()
|
||||
|
||||
with open(export_dir / "patterns.json", 'w') as f:
|
||||
json.dump(shadow_learner.db.to_dict(), f, indent=2)
|
||||
|
||||
# Export logs
|
||||
logs_dir = Path(".claude_hooks/logs")
|
||||
if logs_dir.exists():
|
||||
import shutil
|
||||
shutil.copytree(logs_dir, export_dir / "logs", dirs_exist_ok=True)
|
||||
|
||||
print(f"✅ Data exported to {export_dir}")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main CLI entry point"""
|
||||
parser = argparse.ArgumentParser(description="Claude Code Hooks CLI")
|
||||
|
||||
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
||||
|
||||
# List backups
|
||||
subparsers.add_parser("list-backups", help="List available backups")
|
||||
|
||||
# Show session status
|
||||
subparsers.add_parser("status", help="Show current session status")
|
||||
|
||||
# Show patterns
|
||||
subparsers.add_parser("patterns", help="Show learned patterns")
|
||||
|
||||
# Clear patterns
|
||||
subparsers.add_parser("clear-patterns", help="Clear all learned patterns")
|
||||
|
||||
# Export data
|
||||
subparsers.add_parser("export", help="Export all hook data")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
return
|
||||
|
||||
try:
|
||||
if args.command == "list-backups":
|
||||
list_backups()
|
||||
elif args.command == "status":
|
||||
show_session_status()
|
||||
elif args.command == "patterns":
|
||||
show_patterns()
|
||||
elif args.command == "clear-patterns":
|
||||
clear_patterns()
|
||||
elif args.command == "export":
|
||||
export_data()
|
||||
else:
|
||||
print(f"Unknown command: {args.command}")
|
||||
parser.print_help()
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
321
lib/context_monitor.py
Normal file
321
lib/context_monitor.py
Normal file
|
|
@ -0,0 +1,321 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Context Monitor - Token estimation and backup trigger system"""
|
||||
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
try:
|
||||
from .models import BackupDecision
|
||||
except ImportError:
|
||||
from models import BackupDecision
|
||||
|
||||
|
||||
class ContextMonitor:
|
||||
"""Monitors conversation context and predicts token usage"""
|
||||
|
||||
def __init__(self, storage_path: str = ".claude_hooks"):
|
||||
self.storage_path = Path(storage_path)
|
||||
self.storage_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
self.session_start = datetime.now()
|
||||
self.prompt_count = 0
|
||||
self.estimated_tokens = 0
|
||||
self.tool_executions = 0
|
||||
self.file_operations = 0
|
||||
|
||||
# Token estimation constants (conservative estimates)
|
||||
self.TOKENS_PER_CHAR = 0.25 # Average for English text
|
||||
self.TOOL_OVERHEAD = 200 # Tokens per tool call
|
||||
self.SYSTEM_OVERHEAD = 500 # Base conversation overhead
|
||||
self.MAX_CONTEXT = 200000 # Claude's context limit
|
||||
|
||||
# Backup thresholds
|
||||
self.backup_threshold = 0.85
|
||||
self.emergency_threshold = 0.95
|
||||
|
||||
# Error tracking
|
||||
self.estimation_errors = 0
|
||||
self.max_errors = 5
|
||||
self._last_good_estimate = 0.5
|
||||
|
||||
# Load previous session state if available
|
||||
self._load_session_state()
|
||||
|
||||
def estimate_prompt_tokens(self, prompt_data: Dict[str, Any]) -> int:
|
||||
"""Estimate tokens in user prompt"""
|
||||
try:
|
||||
prompt_text = prompt_data.get("prompt", "")
|
||||
|
||||
# Basic character count estimation
|
||||
base_tokens = len(prompt_text) * self.TOKENS_PER_CHAR
|
||||
|
||||
# Add overhead for system prompts, context, etc.
|
||||
overhead_tokens = self.SYSTEM_OVERHEAD
|
||||
|
||||
return int(base_tokens + overhead_tokens)
|
||||
|
||||
except Exception:
|
||||
# Fallback estimation
|
||||
return 1000
|
||||
|
||||
def estimate_conversation_tokens(self) -> int:
|
||||
"""Estimate total conversation tokens"""
|
||||
try:
|
||||
# Base conversation context
|
||||
base_tokens = self.estimated_tokens
|
||||
|
||||
# Add tool execution overhead
|
||||
tool_tokens = self.tool_executions * self.TOOL_OVERHEAD
|
||||
|
||||
# Add file operation overhead (file contents in context)
|
||||
file_tokens = self.file_operations * 1000 # Average file size
|
||||
|
||||
# Conversation history grows over time
|
||||
history_tokens = self.prompt_count * 300 # Average response size
|
||||
|
||||
total = base_tokens + tool_tokens + file_tokens + history_tokens
|
||||
|
||||
return min(total, self.MAX_CONTEXT)
|
||||
|
||||
except Exception:
|
||||
return self._handle_estimation_failure()
|
||||
|
||||
def get_context_usage_ratio(self) -> float:
|
||||
"""Get estimated context usage as ratio (0.0 to 1.0)"""
|
||||
try:
|
||||
estimated = self.estimate_conversation_tokens()
|
||||
ratio = min(1.0, estimated / self.MAX_CONTEXT)
|
||||
|
||||
# Reset error counter on success
|
||||
self.estimation_errors = 0
|
||||
self._last_good_estimate = ratio
|
||||
|
||||
return ratio
|
||||
|
||||
except Exception:
|
||||
self.estimation_errors += 1
|
||||
|
||||
# Too many errors - use conservative fallback
|
||||
if self.estimation_errors >= self.max_errors:
|
||||
return 0.7 # Conservative threshold
|
||||
|
||||
# Single error - use last known good value
|
||||
return self._last_good_estimate
|
||||
|
||||
def should_trigger_backup(self, threshold: Optional[float] = None) -> bool:
|
||||
"""Check if backup should be triggered"""
|
||||
try:
|
||||
if threshold is None:
|
||||
threshold = self.backup_threshold
|
||||
|
||||
usage = self.get_context_usage_ratio()
|
||||
|
||||
# Edge case: Very early in session
|
||||
if self.prompt_count < 2:
|
||||
return False
|
||||
|
||||
# Edge case: Already near context limit
|
||||
if usage > self.emergency_threshold:
|
||||
# Emergency backup - don't wait for other conditions
|
||||
return True
|
||||
|
||||
# Session duration factor
|
||||
session_hours = (datetime.now() - self.session_start).total_seconds() / 3600
|
||||
complexity_factor = (self.tool_executions + self.file_operations) / 20
|
||||
|
||||
# Trigger earlier for complex sessions
|
||||
adjusted_threshold = threshold - (complexity_factor * 0.1)
|
||||
|
||||
# Multiple trigger conditions
|
||||
return (
|
||||
usage > adjusted_threshold or
|
||||
session_hours > 2.0 or
|
||||
(usage > 0.7 and session_hours > 1.0)
|
||||
)
|
||||
|
||||
except Exception:
|
||||
# When in doubt, backup (better safe than sorry)
|
||||
return True
|
||||
|
||||
def update_from_prompt(self, prompt_data: Dict[str, Any]):
|
||||
"""Update estimates when user submits prompt"""
|
||||
try:
|
||||
self.prompt_count += 1
|
||||
prompt_tokens = self.estimate_prompt_tokens(prompt_data)
|
||||
self.estimated_tokens += prompt_tokens
|
||||
|
||||
# Save state periodically
|
||||
if self.prompt_count % 5 == 0:
|
||||
self._save_session_state()
|
||||
|
||||
except Exception:
|
||||
pass # Don't let tracking errors break the system
|
||||
|
||||
def update_from_tool_use(self, tool_data: Dict[str, Any]):
|
||||
"""Update estimates when tools are used"""
|
||||
try:
|
||||
self.tool_executions += 1
|
||||
|
||||
tool_name = tool_data.get("tool", "")
|
||||
|
||||
# File operations add content to context
|
||||
if tool_name in ["Read", "Edit", "Write", "Glob", "MultiEdit"]:
|
||||
self.file_operations += 1
|
||||
|
||||
# Large outputs add to context
|
||||
parameters = tool_data.get("parameters", {})
|
||||
if "file_path" in parameters:
|
||||
self.estimated_tokens += 500 # Estimated file content
|
||||
|
||||
# Save state periodically
|
||||
if self.tool_executions % 10 == 0:
|
||||
self._save_session_state()
|
||||
|
||||
except Exception:
|
||||
pass # Don't let tracking errors break the system
|
||||
|
||||
def check_backup_triggers(self, hook_event: str, data: Dict[str, Any]) -> BackupDecision:
|
||||
"""Check all backup trigger conditions"""
|
||||
try:
|
||||
# Context-based triggers
|
||||
if self.should_trigger_backup():
|
||||
usage = self.get_context_usage_ratio()
|
||||
urgency = "high" if usage > self.emergency_threshold else "medium"
|
||||
|
||||
return BackupDecision(
|
||||
should_backup=True,
|
||||
reason="context_threshold",
|
||||
urgency=urgency,
|
||||
metadata={"usage_ratio": usage}
|
||||
)
|
||||
|
||||
# Activity-based triggers
|
||||
if self._should_backup_by_activity():
|
||||
return BackupDecision(
|
||||
should_backup=True,
|
||||
reason="activity_threshold",
|
||||
urgency="medium"
|
||||
)
|
||||
|
||||
# Critical operation triggers
|
||||
if self._is_critical_operation(data):
|
||||
return BackupDecision(
|
||||
should_backup=True,
|
||||
reason="critical_operation",
|
||||
urgency="high"
|
||||
)
|
||||
|
||||
return BackupDecision(should_backup=False, reason="no_trigger")
|
||||
|
||||
except Exception:
|
||||
# If trigger checking fails, err on side of safety
|
||||
return BackupDecision(
|
||||
should_backup=True,
|
||||
reason="trigger_check_failed",
|
||||
urgency="medium"
|
||||
)
|
||||
|
||||
def _should_backup_by_activity(self) -> bool:
|
||||
"""Activity-based backup triggers"""
|
||||
# Backup after significant file modifications
|
||||
if (self.file_operations % 10 == 0 and self.file_operations > 0):
|
||||
return True
|
||||
|
||||
# Backup after many tool executions
|
||||
if (self.tool_executions % 25 == 0 and self.tool_executions > 0):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _is_critical_operation(self, data: Dict[str, Any]) -> bool:
|
||||
"""Detect operations that should trigger immediate backup"""
|
||||
tool = data.get("tool", "")
|
||||
params = data.get("parameters", {})
|
||||
|
||||
# Git operations
|
||||
if tool == "Bash":
|
||||
command = params.get("command", "").lower()
|
||||
if any(git_cmd in command for git_cmd in ["git commit", "git push", "git merge"]):
|
||||
return True
|
||||
|
||||
# Package installations
|
||||
if any(pkg_cmd in command for pkg_cmd in ["npm install", "pip install", "cargo install"]):
|
||||
return True
|
||||
|
||||
# Major file operations
|
||||
if tool in ["Write", "MultiEdit"]:
|
||||
content = params.get("content", "")
|
||||
if len(content) > 5000: # Large file changes
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _handle_estimation_failure(self) -> int:
|
||||
"""Fallback estimation when primary method fails"""
|
||||
# Method 1: Time-based estimation
|
||||
session_duration = (datetime.now() - self.session_start).total_seconds() / 3600
|
||||
if session_duration > 1.0: # 1 hour = likely high usage
|
||||
return int(self.MAX_CONTEXT * 0.8)
|
||||
|
||||
# Method 2: Activity-based estimation
|
||||
total_activity = self.tool_executions + self.file_operations
|
||||
if total_activity > 50: # High activity = likely high context
|
||||
return int(self.MAX_CONTEXT * 0.75)
|
||||
|
||||
# Method 3: Conservative default
|
||||
return int(self.MAX_CONTEXT * 0.5)
|
||||
|
||||
def _save_session_state(self):
|
||||
"""Save current session state to disk"""
|
||||
try:
|
||||
state_file = self.storage_path / "session_state.json"
|
||||
|
||||
state = {
|
||||
"session_start": self.session_start.isoformat(),
|
||||
"prompt_count": self.prompt_count,
|
||||
"estimated_tokens": self.estimated_tokens,
|
||||
"tool_executions": self.tool_executions,
|
||||
"file_operations": self.file_operations,
|
||||
"last_updated": datetime.now().isoformat()
|
||||
}
|
||||
|
||||
with open(state_file, 'w') as f:
|
||||
json.dump(state, f, indent=2)
|
||||
|
||||
except Exception:
|
||||
pass # Don't let state saving errors break the system
|
||||
|
||||
def _load_session_state(self):
|
||||
"""Load previous session state if available"""
|
||||
try:
|
||||
state_file = self.storage_path / "session_state.json"
|
||||
|
||||
if state_file.exists():
|
||||
with open(state_file, 'r') as f:
|
||||
state = json.load(f)
|
||||
|
||||
# Only load if session is recent (within last hour)
|
||||
last_updated = datetime.fromisoformat(state["last_updated"])
|
||||
if datetime.now() - last_updated < timedelta(hours=1):
|
||||
self.prompt_count = state.get("prompt_count", 0)
|
||||
self.estimated_tokens = state.get("estimated_tokens", 0)
|
||||
self.tool_executions = state.get("tool_executions", 0)
|
||||
self.file_operations = state.get("file_operations", 0)
|
||||
|
||||
except Exception:
|
||||
pass # If loading fails, start fresh
|
||||
|
||||
def get_session_summary(self) -> Dict[str, Any]:
|
||||
"""Get current session summary"""
|
||||
return {
|
||||
"session_duration": str(datetime.now() - self.session_start),
|
||||
"prompt_count": self.prompt_count,
|
||||
"tool_executions": self.tool_executions,
|
||||
"file_operations": self.file_operations,
|
||||
"estimated_tokens": self.estimate_conversation_tokens(),
|
||||
"context_usage_ratio": self.get_context_usage_ratio(),
|
||||
"should_backup": self.should_trigger_backup()
|
||||
}
|
||||
198
lib/models.py
Normal file
198
lib/models.py
Normal file
|
|
@ -0,0 +1,198 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Data models for Claude Code Hooks system"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Optional, Any
|
||||
import json
|
||||
|
||||
|
||||
@dataclass
|
||||
class ToolExecution:
|
||||
"""Single tool execution record"""
|
||||
timestamp: datetime
|
||||
tool: str
|
||||
parameters: Dict[str, Any]
|
||||
success: bool
|
||||
error_message: Optional[str] = None
|
||||
execution_time: float = 0.0
|
||||
context: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"timestamp": self.timestamp.isoformat(),
|
||||
"tool": self.tool,
|
||||
"parameters": self.parameters,
|
||||
"success": self.success,
|
||||
"error_message": self.error_message,
|
||||
"execution_time": self.execution_time,
|
||||
"context": self.context
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'ToolExecution':
|
||||
return cls(
|
||||
timestamp=datetime.fromisoformat(data["timestamp"]),
|
||||
tool=data["tool"],
|
||||
parameters=data["parameters"],
|
||||
success=data["success"],
|
||||
error_message=data.get("error_message"),
|
||||
execution_time=data.get("execution_time", 0.0),
|
||||
context=data.get("context", {})
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Pattern:
|
||||
"""Learned pattern with confidence scoring"""
|
||||
pattern_id: str
|
||||
pattern_type: str # "command_failure", "tool_sequence", "context_error"
|
||||
trigger: Dict[str, Any] # What triggers this pattern
|
||||
prediction: Dict[str, Any] # What we predict will happen
|
||||
confidence: float # 0.0 to 1.0
|
||||
evidence_count: int # How many times we've seen this
|
||||
last_seen: datetime
|
||||
success_rate: float = 0.0
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"pattern_id": self.pattern_id,
|
||||
"pattern_type": self.pattern_type,
|
||||
"trigger": self.trigger,
|
||||
"prediction": self.prediction,
|
||||
"confidence": self.confidence,
|
||||
"evidence_count": self.evidence_count,
|
||||
"last_seen": self.last_seen.isoformat(),
|
||||
"success_rate": self.success_rate
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'Pattern':
|
||||
return cls(
|
||||
pattern_id=data["pattern_id"],
|
||||
pattern_type=data["pattern_type"],
|
||||
trigger=data["trigger"],
|
||||
prediction=data["prediction"],
|
||||
confidence=data["confidence"],
|
||||
evidence_count=data["evidence_count"],
|
||||
last_seen=datetime.fromisoformat(data["last_seen"]),
|
||||
success_rate=data.get("success_rate", 0.0)
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class HookResult:
|
||||
"""Result of hook execution"""
|
||||
allow: bool
|
||||
message: str = ""
|
||||
warning: bool = False
|
||||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@classmethod
|
||||
def success(cls, message: str = "Operation allowed") -> 'HookResult':
|
||||
return cls(allow=True, message=message)
|
||||
|
||||
@classmethod
|
||||
def blocked(cls, reason: str) -> 'HookResult':
|
||||
return cls(allow=False, message=reason)
|
||||
|
||||
@classmethod
|
||||
def allow_with_warning(cls, warning: str) -> 'HookResult':
|
||||
return cls(allow=True, message=warning, warning=True)
|
||||
|
||||
def to_claude_response(self) -> Dict[str, Any]:
|
||||
"""Convert to Claude Code hook response format"""
|
||||
response = {
|
||||
"allow": self.allow,
|
||||
"message": self.message
|
||||
}
|
||||
if self.metadata:
|
||||
response.update(self.metadata)
|
||||
return response
|
||||
|
||||
|
||||
@dataclass
|
||||
class ValidationResult:
|
||||
"""Result of validation operations"""
|
||||
allowed: bool
|
||||
reason: str = ""
|
||||
severity: str = "info" # info, warning, medium, high, critical
|
||||
suggestions: List[str] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def is_critical(self) -> bool:
|
||||
return self.severity == "critical"
|
||||
|
||||
@property
|
||||
def is_blocking(self) -> bool:
|
||||
return not self.allowed
|
||||
|
||||
|
||||
@dataclass
|
||||
class BackupDecision:
|
||||
"""Decision about whether to trigger backup"""
|
||||
should_backup: bool
|
||||
reason: str
|
||||
urgency: str = "medium" # low, medium, high
|
||||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
class BackupResult:
|
||||
"""Result of backup operation"""
|
||||
success: bool
|
||||
backup_id: str = ""
|
||||
backup_path: str = ""
|
||||
error: str = ""
|
||||
git_success: bool = False
|
||||
fallback_performed: bool = False
|
||||
components: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
class GitBackupResult:
|
||||
"""Result of git backup operation"""
|
||||
success: bool
|
||||
commit_id: str = ""
|
||||
message: str = ""
|
||||
error: str = ""
|
||||
|
||||
|
||||
class PatternDatabase:
|
||||
"""Fast lookup database for learned patterns"""
|
||||
|
||||
def __init__(self):
|
||||
self.command_patterns: Dict[str, Pattern] = {}
|
||||
self.sequence_patterns: List[Pattern] = []
|
||||
self.context_patterns: Dict[str, Pattern] = {}
|
||||
self.execution_history: List[ToolExecution] = []
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"command_patterns": {k: v.to_dict() for k, v in self.command_patterns.items()},
|
||||
"sequence_patterns": [p.to_dict() for p in self.sequence_patterns],
|
||||
"context_patterns": {k: v.to_dict() for k, v in self.context_patterns.items()},
|
||||
"execution_history": [e.to_dict() for e in self.execution_history[-100:]] # Keep last 100
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'PatternDatabase':
|
||||
db = cls()
|
||||
|
||||
# Load command patterns
|
||||
for k, v in data.get("command_patterns", {}).items():
|
||||
db.command_patterns[k] = Pattern.from_dict(v)
|
||||
|
||||
# Load sequence patterns
|
||||
for p in data.get("sequence_patterns", []):
|
||||
db.sequence_patterns.append(Pattern.from_dict(p))
|
||||
|
||||
# Load context patterns
|
||||
for k, v in data.get("context_patterns", {}).items():
|
||||
db.context_patterns[k] = Pattern.from_dict(v)
|
||||
|
||||
# Load execution history
|
||||
for e in data.get("execution_history", []):
|
||||
db.execution_history.append(ToolExecution.from_dict(e))
|
||||
|
||||
return db
|
||||
348
lib/session_state.py
Normal file
348
lib/session_state.py
Normal file
|
|
@ -0,0 +1,348 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Session State Manager - Persistent session state and continuity"""
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Any, Set
|
||||
|
||||
try:
|
||||
from .models import ToolExecution
|
||||
except ImportError:
|
||||
from models import ToolExecution
|
||||
|
||||
|
||||
class SessionStateManager:
|
||||
"""Manages persistent session state across Claude interactions"""
|
||||
|
||||
def __init__(self, state_dir: str = ".claude_hooks"):
|
||||
self.state_dir = Path(state_dir)
|
||||
self.state_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
self.state_file = self.state_dir / "session_state.json"
|
||||
self.todos_file = Path("ACTIVE_TODOS.md")
|
||||
self.last_session_file = Path("LAST_SESSION.md")
|
||||
|
||||
# Initialize session
|
||||
self.session_id = str(uuid.uuid4())[:8]
|
||||
self.current_state = self._load_or_create_state()
|
||||
|
||||
def _load_or_create_state(self) -> Dict[str, Any]:
|
||||
"""Load existing state or create new session state"""
|
||||
try:
|
||||
if self.state_file.exists():
|
||||
with open(self.state_file, 'r') as f:
|
||||
state = json.load(f)
|
||||
|
||||
# Check if this is a continuation of recent session
|
||||
last_activity = datetime.fromisoformat(state.get("last_activity", "1970-01-01"))
|
||||
if (datetime.now() - last_activity).total_seconds() < 3600: # Within 1 hour
|
||||
# Continue existing session
|
||||
return state
|
||||
|
||||
# Create new session
|
||||
return self._create_new_session()
|
||||
|
||||
except Exception:
|
||||
# If loading fails, create new session
|
||||
return self._create_new_session()
|
||||
|
||||
def _create_new_session(self) -> Dict[str, Any]:
|
||||
"""Create new session state"""
|
||||
return {
|
||||
"session_id": self.session_id,
|
||||
"start_time": datetime.now().isoformat(),
|
||||
"last_activity": datetime.now().isoformat(),
|
||||
"modified_files": [],
|
||||
"commands_executed": [],
|
||||
"tool_usage": {},
|
||||
"backup_history": [],
|
||||
"todos": [],
|
||||
"context_snapshots": []
|
||||
}
|
||||
|
||||
def update_from_tool_use(self, tool_data: Dict[str, Any]):
|
||||
"""Update session state from tool usage"""
|
||||
try:
|
||||
tool = tool_data.get("tool", "")
|
||||
params = tool_data.get("parameters", {})
|
||||
timestamp = datetime.now().isoformat()
|
||||
|
||||
# Track file modifications
|
||||
if tool in ["Edit", "Write", "MultiEdit"]:
|
||||
file_path = params.get("file_path", "")
|
||||
if file_path and file_path not in self.current_state["modified_files"]:
|
||||
self.current_state["modified_files"].append(file_path)
|
||||
|
||||
# Track commands executed
|
||||
if tool == "Bash":
|
||||
command = params.get("command", "")
|
||||
if command:
|
||||
self.current_state["commands_executed"].append({
|
||||
"command": command,
|
||||
"timestamp": timestamp
|
||||
})
|
||||
|
||||
# Keep only last 50 commands
|
||||
if len(self.current_state["commands_executed"]) > 50:
|
||||
self.current_state["commands_executed"] = self.current_state["commands_executed"][-50:]
|
||||
|
||||
# Track tool usage statistics
|
||||
self.current_state["tool_usage"][tool] = self.current_state["tool_usage"].get(tool, 0) + 1
|
||||
self.current_state["last_activity"] = timestamp
|
||||
|
||||
# Save state periodically
|
||||
self._save_state()
|
||||
|
||||
except Exception:
|
||||
pass # Don't let state tracking errors break the system
|
||||
|
||||
def add_backup(self, backup_id: str, backup_info: Dict[str, Any]):
|
||||
"""Record backup in session history"""
|
||||
try:
|
||||
backup_record = {
|
||||
"backup_id": backup_id,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"reason": backup_info.get("reason", "unknown"),
|
||||
"success": backup_info.get("success", False)
|
||||
}
|
||||
|
||||
self.current_state["backup_history"].append(backup_record)
|
||||
|
||||
# Keep only last 10 backups
|
||||
if len(self.current_state["backup_history"]) > 10:
|
||||
self.current_state["backup_history"] = self.current_state["backup_history"][-10:]
|
||||
|
||||
self._save_state()
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def add_context_snapshot(self, context_data: Dict[str, Any]):
|
||||
"""Add context snapshot for recovery"""
|
||||
try:
|
||||
snapshot = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"context_ratio": context_data.get("usage_ratio", 0.0),
|
||||
"prompt_count": context_data.get("prompt_count", 0),
|
||||
"tool_count": context_data.get("tool_executions", 0)
|
||||
}
|
||||
|
||||
self.current_state["context_snapshots"].append(snapshot)
|
||||
|
||||
# Keep only last 20 snapshots
|
||||
if len(self.current_state["context_snapshots"]) > 20:
|
||||
self.current_state["context_snapshots"] = self.current_state["context_snapshots"][-20:]
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def update_todos(self, todos: List[Dict[str, Any]]):
|
||||
"""Update active todos list"""
|
||||
try:
|
||||
self.current_state["todos"] = todos
|
||||
self._save_state()
|
||||
self._update_todos_file()
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def get_session_summary(self) -> Dict[str, Any]:
|
||||
"""Generate comprehensive session summary"""
|
||||
try:
|
||||
return {
|
||||
"session_id": self.current_state.get("session_id", "unknown"),
|
||||
"start_time": self.current_state.get("start_time", "unknown"),
|
||||
"last_activity": self.current_state.get("last_activity", "unknown"),
|
||||
"modified_files": self.current_state.get("modified_files", []),
|
||||
"tool_usage": self.current_state.get("tool_usage", {}),
|
||||
"commands_executed": self.current_state.get("commands_executed", []),
|
||||
"backup_history": self.current_state.get("backup_history", []),
|
||||
"todos": self.current_state.get("todos", []),
|
||||
"session_stats": self._calculate_session_stats()
|
||||
}
|
||||
except Exception:
|
||||
return {"error": "Failed to generate session summary"}
|
||||
|
||||
def _calculate_session_stats(self) -> Dict[str, Any]:
|
||||
"""Calculate session statistics"""
|
||||
try:
|
||||
total_tools = sum(self.current_state.get("tool_usage", {}).values())
|
||||
total_commands = len(self.current_state.get("commands_executed", []))
|
||||
total_files = len(self.current_state.get("modified_files", []))
|
||||
|
||||
start_time = datetime.fromisoformat(self.current_state.get("start_time", datetime.now().isoformat()))
|
||||
duration = datetime.now() - start_time
|
||||
|
||||
return {
|
||||
"duration_minutes": round(duration.total_seconds() / 60, 1),
|
||||
"total_tool_calls": total_tools,
|
||||
"total_commands": total_commands,
|
||||
"total_files_modified": total_files,
|
||||
"most_used_tools": self._get_top_tools(3)
|
||||
}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
def _get_top_tools(self, count: int) -> List[Dict[str, Any]]:
|
||||
"""Get most frequently used tools"""
|
||||
try:
|
||||
tool_usage = self.current_state.get("tool_usage", {})
|
||||
sorted_tools = sorted(tool_usage.items(), key=lambda x: x[1], reverse=True)
|
||||
return [{"tool": tool, "count": usage} for tool, usage in sorted_tools[:count]]
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
def create_continuation_docs(self):
|
||||
"""Create LAST_SESSION.md and ACTIVE_TODOS.md"""
|
||||
try:
|
||||
self._create_last_session_doc()
|
||||
self._update_todos_file()
|
||||
except Exception:
|
||||
pass # Don't let doc creation errors break the system
|
||||
|
||||
def _create_last_session_doc(self):
|
||||
"""Create LAST_SESSION.md with session summary"""
|
||||
try:
|
||||
summary = self.get_session_summary()
|
||||
|
||||
content = f"""# Last Claude Session Summary
|
||||
|
||||
**Session ID**: {summary['session_id']}
|
||||
**Duration**: {summary['start_time']} → {summary['last_activity']}
|
||||
**Session Length**: {summary.get('session_stats', {}).get('duration_minutes', 0)} minutes
|
||||
|
||||
## Files Modified ({len(summary['modified_files'])})
|
||||
"""
|
||||
|
||||
for file_path in summary['modified_files']:
|
||||
content += f"- {file_path}\n"
|
||||
|
||||
content += f"\n## Tools Used ({summary.get('session_stats', {}).get('total_tool_calls', 0)} total)\n"
|
||||
|
||||
for tool, count in summary['tool_usage'].items():
|
||||
content += f"- {tool}: {count} times\n"
|
||||
|
||||
content += f"\n## Recent Commands ({len(summary['commands_executed'])})\n"
|
||||
|
||||
# Show last 10 commands
|
||||
recent_commands = summary['commands_executed'][-10:]
|
||||
for cmd_info in recent_commands:
|
||||
timestamp = cmd_info['timestamp'][:19] # Remove microseconds
|
||||
content += f"- `{cmd_info['command']}` ({timestamp})\n"
|
||||
|
||||
content += f"\n## Backup History\n"
|
||||
|
||||
for backup in summary['backup_history']:
|
||||
status = "✅" if backup['success'] else "❌"
|
||||
content += f"- {status} {backup['backup_id']} - {backup['reason']} ({backup['timestamp'][:19]})\n"
|
||||
|
||||
content += f"""
|
||||
## To Continue This Session
|
||||
|
||||
1. **Review Modified Files**: Check the files listed above for your recent changes
|
||||
2. **Check Active Tasks**: Review `ACTIVE_TODOS.md` for pending work
|
||||
3. **Restore Context**: Reference the commands and tools used above
|
||||
4. **Use Backups**: If needed, restore from backup using `claude-hooks restore {summary['backup_history'][-1]['backup_id'] if summary['backup_history'] else 'latest'}`
|
||||
|
||||
## Quick Commands
|
||||
|
||||
```bash
|
||||
# View current project status
|
||||
git status
|
||||
|
||||
# Check for any uncommitted changes
|
||||
git diff
|
||||
|
||||
# List available backups
|
||||
claude-hooks list-backups
|
||||
|
||||
# Continue with active todos
|
||||
cat ACTIVE_TODOS.md
|
||||
```
|
||||
"""
|
||||
|
||||
with open(self.last_session_file, 'w') as f:
|
||||
f.write(content)
|
||||
|
||||
except Exception as e:
|
||||
# Create minimal doc on error
|
||||
try:
|
||||
with open(self.last_session_file, 'w') as f:
|
||||
f.write(f"# Last Session\n\nSession ended at {datetime.now().isoformat()}\n\nError creating summary: {e}\n")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _update_todos_file(self):
|
||||
"""Update ACTIVE_TODOS.md file"""
|
||||
try:
|
||||
todos = self.current_state.get("todos", [])
|
||||
|
||||
if not todos:
|
||||
content = """# Active TODOs
|
||||
|
||||
*No active todos. Add some to track your progress!*
|
||||
|
||||
## How to Add TODOs
|
||||
|
||||
Use Claude's TodoWrite tool to manage your task list:
|
||||
- Track progress across sessions
|
||||
- Break down complex tasks
|
||||
- Never lose track of what you're working on
|
||||
"""
|
||||
else:
|
||||
content = f"""# Active TODOs
|
||||
|
||||
*Updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*
|
||||
|
||||
"""
|
||||
|
||||
# Group by status
|
||||
pending_todos = [t for t in todos if t.get('status') == 'pending']
|
||||
in_progress_todos = [t for t in todos if t.get('status') == 'in_progress']
|
||||
completed_todos = [t for t in todos if t.get('status') == 'completed']
|
||||
|
||||
if in_progress_todos:
|
||||
content += "## 🚀 In Progress\n\n"
|
||||
for todo in in_progress_todos:
|
||||
priority = todo.get('priority', 'medium')
|
||||
priority_emoji = {'high': '🔥', 'medium': '⭐', 'low': '📝'}.get(priority, '⭐')
|
||||
content += f"- {priority_emoji} {todo.get('content', 'Unknown task')}\n"
|
||||
content += "\n"
|
||||
|
||||
if pending_todos:
|
||||
content += "## 📋 Pending\n\n"
|
||||
for todo in pending_todos:
|
||||
priority = todo.get('priority', 'medium')
|
||||
priority_emoji = {'high': '🔥', 'medium': '⭐', 'low': '📝'}.get(priority, '⭐')
|
||||
content += f"- {priority_emoji} {todo.get('content', 'Unknown task')}\n"
|
||||
content += "\n"
|
||||
|
||||
if completed_todos:
|
||||
content += "## ✅ Completed\n\n"
|
||||
for todo in completed_todos[-5:]: # Show last 5 completed
|
||||
content += f"- ✅ {todo.get('content', 'Unknown task')}\n"
|
||||
content += "\n"
|
||||
|
||||
with open(self.todos_file, 'w') as f:
|
||||
f.write(content)
|
||||
|
||||
except Exception:
|
||||
pass # Don't let todo file creation break the system
|
||||
|
||||
def _save_state(self):
|
||||
"""Save current state to disk"""
|
||||
try:
|
||||
with open(self.state_file, 'w') as f:
|
||||
json.dump(self.current_state, f, indent=2)
|
||||
except Exception:
|
||||
pass # Don't let state saving errors break the system
|
||||
|
||||
def cleanup_session(self):
|
||||
"""Clean up session and create final documentation"""
|
||||
try:
|
||||
self.create_continuation_docs()
|
||||
self._save_state()
|
||||
except Exception:
|
||||
pass
|
||||
395
lib/shadow_learner.py
Normal file
395
lib/shadow_learner.py
Normal file
|
|
@ -0,0 +1,395 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Shadow Learner - Pattern learning and prediction system"""
|
||||
|
||||
import json
|
||||
import math
|
||||
import time
|
||||
import difflib
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Any
|
||||
from cachetools import TTLCache, LRUCache
|
||||
|
||||
try:
|
||||
from .models import Pattern, ToolExecution, PatternDatabase, ValidationResult
|
||||
except ImportError:
|
||||
from models import Pattern, ToolExecution, PatternDatabase, ValidationResult
|
||||
|
||||
|
||||
class ConfidenceCalculator:
|
||||
"""Calculate confidence scores for learned patterns"""
|
||||
|
||||
@staticmethod
|
||||
def calculate_command_confidence(success_count: int, failure_count: int,
|
||||
recency_factor: float) -> float:
|
||||
"""Calculate confidence for command failure patterns"""
|
||||
total_attempts = success_count + failure_count
|
||||
if total_attempts == 0:
|
||||
return 0.0
|
||||
|
||||
# Base confidence from failure rate
|
||||
failure_rate = failure_count / total_attempts
|
||||
|
||||
# Sample size adjustment (more data = more confidence)
|
||||
sample_factor = min(1.0, total_attempts / 10.0) # Plateau at 10 samples
|
||||
|
||||
# Time decay (recent failures are more relevant)
|
||||
confidence = (failure_rate * sample_factor * (0.5 + 0.5 * recency_factor))
|
||||
|
||||
return min(0.99, max(0.1, confidence)) # Clamp between 0.1 and 0.99
|
||||
|
||||
@staticmethod
|
||||
def calculate_sequence_confidence(successful_sequences: int,
|
||||
total_sequences: int) -> float:
|
||||
"""Calculate confidence for tool sequence patterns"""
|
||||
if total_sequences == 0:
|
||||
return 0.0
|
||||
|
||||
success_rate = successful_sequences / total_sequences
|
||||
sample_factor = min(1.0, total_sequences / 5.0)
|
||||
|
||||
return success_rate * sample_factor
|
||||
|
||||
|
||||
class PatternMatcher:
|
||||
"""Advanced pattern matching with fuzzy logic"""
|
||||
|
||||
def __init__(self, db: PatternDatabase):
|
||||
self.db = db
|
||||
|
||||
def fuzzy_command_match(self, command: str, threshold: float = 0.8) -> List[Pattern]:
|
||||
"""Find similar command patterns using fuzzy matching"""
|
||||
cmd_tokens = command.lower().split()
|
||||
if not cmd_tokens:
|
||||
return []
|
||||
|
||||
base_cmd = cmd_tokens[0]
|
||||
matches = []
|
||||
|
||||
for pattern in self.db.command_patterns.values():
|
||||
pattern_cmd = pattern.trigger.get("command", "").lower()
|
||||
|
||||
# Exact match
|
||||
if pattern_cmd == base_cmd:
|
||||
matches.append(pattern)
|
||||
# Fuzzy match on command name
|
||||
elif difflib.SequenceMatcher(None, pattern_cmd, base_cmd).ratio() > threshold:
|
||||
matches.append(pattern)
|
||||
# Partial match (e.g., "pip3" matches "pip install")
|
||||
elif any(pattern_cmd in token for token in cmd_tokens):
|
||||
matches.append(pattern)
|
||||
|
||||
return sorted(matches, key=lambda p: p.confidence, reverse=True)
|
||||
|
||||
def context_pattern_match(self, current_context: Dict[str, Any]) -> List[Pattern]:
|
||||
"""Match patterns based on current context"""
|
||||
matches = []
|
||||
|
||||
for pattern in self.db.context_patterns.values():
|
||||
trigger = pattern.trigger
|
||||
|
||||
# Check if all trigger conditions are met
|
||||
if self._context_matches(current_context, trigger):
|
||||
matches.append(pattern)
|
||||
|
||||
return sorted(matches, key=lambda p: p.confidence, reverse=True)
|
||||
|
||||
def _context_matches(self, current: Dict[str, Any], trigger: Dict[str, Any]) -> bool:
|
||||
"""Check if current context matches trigger conditions"""
|
||||
for key, expected_value in trigger.items():
|
||||
if key not in current:
|
||||
return False
|
||||
|
||||
current_value = current[key]
|
||||
|
||||
# Handle different value types
|
||||
if isinstance(expected_value, str) and isinstance(current_value, str):
|
||||
if expected_value.lower() not in current_value.lower():
|
||||
return False
|
||||
elif expected_value != current_value:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class LearningEngine:
|
||||
"""Core learning algorithms"""
|
||||
|
||||
def __init__(self, db: PatternDatabase):
|
||||
self.db = db
|
||||
self.confidence_calc = ConfidenceCalculator()
|
||||
|
||||
def learn_from_execution(self, execution: ToolExecution):
|
||||
"""Main learning entry point"""
|
||||
|
||||
# Learn command patterns
|
||||
if execution.tool == "Bash":
|
||||
self._learn_command_pattern(execution)
|
||||
|
||||
# Learn tool sequences
|
||||
self._learn_sequence_pattern(execution)
|
||||
|
||||
# Learn context patterns
|
||||
if not execution.success:
|
||||
self._learn_failure_context(execution)
|
||||
|
||||
def _learn_command_pattern(self, execution: ToolExecution):
|
||||
"""Learn from bash command executions"""
|
||||
command = execution.parameters.get("command", "")
|
||||
if not command:
|
||||
return
|
||||
|
||||
base_cmd = command.split()[0]
|
||||
pattern_id = f"cmd_{base_cmd}"
|
||||
|
||||
if pattern_id in self.db.command_patterns:
|
||||
pattern = self.db.command_patterns[pattern_id]
|
||||
|
||||
# Update statistics
|
||||
if execution.success:
|
||||
pattern.prediction["success_count"] = pattern.prediction.get("success_count", 0) + 1
|
||||
else:
|
||||
pattern.prediction["failure_count"] = pattern.prediction.get("failure_count", 0) + 1
|
||||
|
||||
# Recalculate confidence
|
||||
recency = self._calculate_recency(execution.timestamp)
|
||||
pattern.confidence = self.confidence_calc.calculate_command_confidence(
|
||||
pattern.prediction.get("success_count", 0),
|
||||
pattern.prediction.get("failure_count", 0),
|
||||
recency
|
||||
)
|
||||
pattern.last_seen = execution.timestamp
|
||||
pattern.evidence_count += 1
|
||||
|
||||
else:
|
||||
# Create new pattern
|
||||
self.db.command_patterns[pattern_id] = Pattern(
|
||||
pattern_id=pattern_id,
|
||||
pattern_type="command_execution",
|
||||
trigger={"command": base_cmd},
|
||||
prediction={
|
||||
"success_count": 1 if execution.success else 0,
|
||||
"failure_count": 0 if execution.success else 1,
|
||||
"common_errors": [execution.error_message] if execution.error_message else []
|
||||
},
|
||||
confidence=0.3, # Start with low confidence
|
||||
evidence_count=1,
|
||||
last_seen=execution.timestamp,
|
||||
success_rate=1.0 if execution.success else 0.0
|
||||
)
|
||||
|
||||
def _learn_sequence_pattern(self, execution: ToolExecution):
|
||||
"""Learn from tool sequence patterns"""
|
||||
# Get recent tool history (last 5 tools)
|
||||
recent_tools = [e.tool for e in self.db.execution_history[-5:]]
|
||||
recent_tools.append(execution.tool)
|
||||
|
||||
# Look for sequences of 2-3 tools
|
||||
for seq_len in [2, 3]:
|
||||
if len(recent_tools) >= seq_len:
|
||||
sequence = tuple(recent_tools[-seq_len:])
|
||||
pattern_id = f"seq_{'_'.join(sequence)}"
|
||||
|
||||
# Update or create sequence pattern
|
||||
# (Simplified implementation - could be expanded)
|
||||
pass
|
||||
|
||||
def _learn_failure_context(self, execution: ToolExecution):
|
||||
"""Learn from failure contexts"""
|
||||
if not execution.error_message:
|
||||
return
|
||||
|
||||
# Extract key error indicators
|
||||
error_key = self._extract_error_key(execution.error_message)
|
||||
if not error_key:
|
||||
return
|
||||
|
||||
pattern_id = f"ctx_error_{error_key}"
|
||||
|
||||
if pattern_id in self.db.context_patterns:
|
||||
pattern = self.db.context_patterns[pattern_id]
|
||||
pattern.evidence_count += 1
|
||||
pattern.last_seen = execution.timestamp
|
||||
# Update confidence based on repeated failures
|
||||
pattern.confidence = min(0.95, pattern.confidence + 0.05)
|
||||
else:
|
||||
# Create new context pattern
|
||||
self.db.context_patterns[pattern_id] = Pattern(
|
||||
pattern_id=pattern_id,
|
||||
pattern_type="context_error",
|
||||
trigger={
|
||||
"tool": execution.tool,
|
||||
"error_type": error_key
|
||||
},
|
||||
prediction={
|
||||
"likely_error": execution.error_message,
|
||||
"suggestions": self._generate_suggestions(execution)
|
||||
},
|
||||
confidence=0.4,
|
||||
evidence_count=1,
|
||||
last_seen=execution.timestamp,
|
||||
success_rate=0.0
|
||||
)
|
||||
|
||||
def _calculate_recency(self, timestamp: datetime) -> float:
|
||||
"""Calculate recency factor (1.0 = very recent, 0.0 = very old)"""
|
||||
now = datetime.now()
|
||||
age_hours = (now - timestamp).total_seconds() / 3600
|
||||
|
||||
# Exponential decay: recent events matter more
|
||||
return max(0.0, math.exp(-age_hours / 24.0)) # 24 hour half-life
|
||||
|
||||
def _extract_error_key(self, error_message: str) -> Optional[str]:
|
||||
"""Extract key error indicators from error messages"""
|
||||
error_message = error_message.lower()
|
||||
|
||||
error_patterns = {
|
||||
"command_not_found": ["command not found", "not found"],
|
||||
"permission_denied": ["permission denied", "access denied"],
|
||||
"file_not_found": ["no such file", "file not found"],
|
||||
"connection_error": ["connection refused", "network unreachable"],
|
||||
"syntax_error": ["syntax error", "invalid syntax"]
|
||||
}
|
||||
|
||||
for error_type, patterns in error_patterns.items():
|
||||
if any(pattern in error_message for pattern in patterns):
|
||||
return error_type
|
||||
|
||||
return None
|
||||
|
||||
def _generate_suggestions(self, execution: ToolExecution) -> List[str]:
|
||||
"""Generate suggestions based on failed execution"""
|
||||
suggestions = []
|
||||
|
||||
if execution.tool == "Bash":
|
||||
command = execution.parameters.get("command", "")
|
||||
if command:
|
||||
base_cmd = command.split()[0]
|
||||
|
||||
# Common command alternatives
|
||||
alternatives = {
|
||||
"pip": ["pip3", "python -m pip", "python3 -m pip"],
|
||||
"python": ["python3"],
|
||||
"node": ["nodejs"],
|
||||
"vim": ["nvim", "nano"],
|
||||
}
|
||||
|
||||
if base_cmd in alternatives:
|
||||
suggestions.extend([f"Try '{alt} {' '.join(command.split()[1:])}'"
|
||||
for alt in alternatives[base_cmd]])
|
||||
|
||||
return suggestions
|
||||
|
||||
|
||||
class PredictionEngine:
|
||||
"""Generate predictions and suggestions"""
|
||||
|
||||
def __init__(self, matcher: PatternMatcher):
|
||||
self.matcher = matcher
|
||||
|
||||
def predict_command_outcome(self, command: str, context: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Predict if a command will succeed and suggest alternatives"""
|
||||
|
||||
# Find matching patterns
|
||||
command_patterns = self.matcher.fuzzy_command_match(command)
|
||||
context_patterns = self.matcher.context_pattern_match(context)
|
||||
|
||||
prediction = {
|
||||
"likely_success": True,
|
||||
"confidence": 0.5,
|
||||
"warnings": [],
|
||||
"suggestions": []
|
||||
}
|
||||
|
||||
# Analyze command patterns
|
||||
for pattern in command_patterns[:3]: # Top 3 matches
|
||||
if pattern.confidence > 0.7:
|
||||
failure_rate = pattern.prediction.get("failure_count", 0) / max(1, pattern.evidence_count)
|
||||
|
||||
if failure_rate > 0.6: # High failure rate
|
||||
prediction["likely_success"] = False
|
||||
prediction["confidence"] = pattern.confidence
|
||||
prediction["warnings"].append(f"Command '{command.split()[0]}' often fails")
|
||||
|
||||
# Add suggestions from pattern
|
||||
suggestions = pattern.prediction.get("suggestions", [])
|
||||
prediction["suggestions"].extend(suggestions)
|
||||
|
||||
return prediction
|
||||
|
||||
|
||||
class ShadowLearner:
|
||||
"""Main shadow learner interface"""
|
||||
|
||||
def __init__(self, storage_path: str = ".claude_hooks/patterns"):
|
||||
self.storage_path = Path(storage_path)
|
||||
self.storage_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
self.db = self._load_database()
|
||||
self.matcher = PatternMatcher(self.db)
|
||||
self.learning_engine = LearningEngine(self.db)
|
||||
self.prediction_engine = PredictionEngine(self.matcher)
|
||||
|
||||
# Performance caches
|
||||
self.prediction_cache = TTLCache(maxsize=1000, ttl=300) # 5-minute cache
|
||||
|
||||
def learn_from_execution(self, execution: ToolExecution):
|
||||
"""Learn from tool execution"""
|
||||
try:
|
||||
self.learning_engine.learn_from_execution(execution)
|
||||
self.db.execution_history.append(execution)
|
||||
|
||||
# Trim history to keep memory usage reasonable
|
||||
if len(self.db.execution_history) > 1000:
|
||||
self.db.execution_history = self.db.execution_history[-500:]
|
||||
|
||||
except Exception as e:
|
||||
# Learning failures shouldn't break the system
|
||||
pass
|
||||
|
||||
def predict_command_outcome(self, command: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
|
||||
"""Predict command outcome with caching"""
|
||||
cache_key = f"cmd_pred:{hash(command)}"
|
||||
|
||||
if cache_key in self.prediction_cache:
|
||||
return self.prediction_cache[cache_key]
|
||||
|
||||
prediction = self.prediction_engine.predict_command_outcome(
|
||||
command, context or {}
|
||||
)
|
||||
|
||||
self.prediction_cache[cache_key] = prediction
|
||||
return prediction
|
||||
|
||||
def save_database(self):
|
||||
"""Save learned patterns to disk"""
|
||||
try:
|
||||
patterns_file = self.storage_path / "patterns.json"
|
||||
backup_file = self.storage_path / "patterns.backup.json"
|
||||
|
||||
# Create backup of existing data
|
||||
if patterns_file.exists():
|
||||
patterns_file.rename(backup_file)
|
||||
|
||||
# Save new data
|
||||
with open(patterns_file, 'w') as f:
|
||||
json.dump(self.db.to_dict(), f, indent=2)
|
||||
|
||||
except Exception as e:
|
||||
# Save failures shouldn't break the system
|
||||
pass
|
||||
|
||||
def _load_database(self) -> PatternDatabase:
|
||||
"""Load patterns database from disk"""
|
||||
patterns_file = self.storage_path / "patterns.json"
|
||||
|
||||
try:
|
||||
if patterns_file.exists():
|
||||
with open(patterns_file, 'r') as f:
|
||||
data = json.load(f)
|
||||
return PatternDatabase.from_dict(data)
|
||||
except Exception:
|
||||
# If loading fails, start with empty database
|
||||
pass
|
||||
|
||||
return PatternDatabase()
|
||||
Loading…
Add table
Add a link
Reference in a new issue