rsp2k/claude-hooks
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Claude hooks auto-backup: manual (backup_20250720_091250)

Browse source
This commit is contained in:
Ryan Malloy 2025-07-20 03:12:51 -06:00
parent 9445e09c48
commit 392833187e
135 changed files with 16151 additions and 3439 deletions
Download patch file Download diff file Expand all files Collapse all files

21
lib/__init__.py
View file

@ -1,21 +0,0 @@
"""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"
]

536
lib/backup-manager.js Normal file
View file

@ -0,0 +1,536 @@
/**
* Backup Manager - Resilient backup execution system
* Node.js implementation
*/
const fs = require('fs-extra');
const path = require('path');
const { spawn } = require('child_process');
class BackupManager {
constructor(projectRoot = '.') {
this.projectRoot = path.resolve(projectRoot);
this.backupDir = path.join(this.projectRoot, '.claude_hooks', 'backups');
fs.ensureDirSync(this.backupDir);
// Backup settings
this.maxBackups = 10;
this.logFile = path.join(this.backupDir, 'backup.log');
}
/**
* Execute backup with comprehensive error handling
*/
async executeBackup(decision, sessionState) {
const backupId = this._generateBackupId();
const backupPath = path.join(this.backupDir, backupId);
try {
// Create backup structure
const backupInfo = await this._createBackupStructure(backupPath, sessionState);
// Git backup (if possible)
const gitResult = await this._attemptGitBackup(backupId, decision.reason);
// File system backup
const fsResult = await this._createFilesystemBackup(backupPath, sessionState);
// Session state backup
const stateResult = await this._backupSessionState(backupPath, sessionState);
// Clean up old backups
await this._cleanupOldBackups();
// Log successful backup
await this._logBackup(backupId, decision, true);
return {
success: true,
backupId,
backupPath,
gitSuccess: gitResult.success,
components: {
git: gitResult,
filesystem: fsResult,
sessionState: stateResult
}
};
} catch (error) {
// Backup failures should never break the session
const fallbackResult = await this._createMinimalBackup(sessionState);
await this._logBackup(backupId, decision, false, error.message);
return {
success: false,
backupId,
error: error.message,
fallbackPerformed: fallbackResult
};
}
}
/**
* Create backup from project path (external API)
*/
async createBackup(projectPath, context = {}, force = false) {
const decision = {
reason: context.trigger || 'manual',
urgency: 'medium',
force
};
const sessionState = {
modifiedFiles: context.modified_files || [],
toolUsage: context.tool_usage || {},
timestamp: new Date().toISOString(),
...context
};
const result = await this.executeBackup(decision, sessionState);
return result.success ? result.backupId : null;
}
/**
* Get backup information
*/
async getBackupInfo(backupId) {
try {
const backupPath = path.join(this.backupDir, backupId);
const metadataFile = path.join(backupPath, 'metadata.json');
if (await fs.pathExists(metadataFile)) {
const metadata = await fs.readJson(metadataFile);
// Add file list if available
const filesDir = path.join(backupPath, 'files');
if (await fs.pathExists(filesDir)) {
const files = await this._getBackupFiles(filesDir);
metadata.files_backed_up = files;
}
return metadata;
}
} catch (error) {
console.error('Error reading backup info:', error.message);
}
return null;
}
/**
* Generate unique backup identifier
*/
_generateBackupId() {
const timestamp = new Date().toISOString()
.replace(/[:-]/g, '')
.replace(/\.\d{3}Z$/, '')
.replace('T', '_');
return `backup_${timestamp}`;
}
/**
* Create basic backup directory structure
*/
async _createBackupStructure(backupPath, sessionState) {
await fs.ensureDir(backupPath);
// Create subdirectories
await fs.ensureDir(path.join(backupPath, 'files'));
await fs.ensureDir(path.join(backupPath, 'logs'));
await fs.ensureDir(path.join(backupPath, 'state'));
// Create backup metadata
const metadata = {
backup_id: path.basename(backupPath),
timestamp: new Date().toISOString(),
session_state: sessionState,
project_root: this.projectRoot
};
await fs.writeJson(path.join(backupPath, 'metadata.json'), metadata, { spaces: 2 });
return metadata;
}
/**
* Attempt git backup with proper error handling
*/
async _attemptGitBackup(backupId, reason) {
try {
// Check if git repo exists
if (!(await fs.pathExists(path.join(this.projectRoot, '.git')))) {
// Initialize repo if none exists
const initResult = await this._runGitCommand(['init']);
if (!initResult.success) {
return {
success: false,
error: `Git init failed: ${initResult.error}`
};
}
}
// Add all changes
const addResult = await this._runGitCommand(['add', '-A']);
if (!addResult.success) {
return {
success: false,
error: `Git add failed: ${addResult.error}`
};
}
// Check if there are changes to commit
const statusResult = await this._runGitCommand(['status', '--porcelain']);
if (statusResult.success && !statusResult.stdout.trim()) {
return {
success: true,
message: 'No changes to commit'
};
}
// Create commit
const commitMsg = `Claude hooks auto-backup: ${reason} (${backupId})`;
const commitResult = await this._runGitCommand(['commit', '-m', commitMsg]);
if (!commitResult.success) {
return {
success: false,
error: `Git commit failed: ${commitResult.error}`
};
}
// Get commit ID
const commitId = await this._getLatestCommit();
return {
success: true,
commitId,
message: `Committed as ${commitId.substring(0, 8)}`
};
} catch (error) {
return {
success: false,
error: `Unexpected git error: ${error.message}`
};
}
}
/**
* Run git command with timeout and error handling
*/
_runGitCommand(args, timeoutMs = 60000) {
return new Promise((resolve) => {
const child = spawn('git', args, {
cwd: this.projectRoot,
stdio: ['pipe', 'pipe', 'pipe']
});
let stdout = '';
let stderr = '';
child.stdout.on('data', (data) => {
stdout += data.toString();
});
child.stderr.on('data', (data) => {
stderr += data.toString();
});
const timeout = setTimeout(() => {
child.kill();
resolve({
success: false,
error: 'Git operation timed out'
});
}, timeoutMs);
child.on('close', (code) => {
clearTimeout(timeout);
resolve({
success: code === 0,
stdout: stdout.trim(),
stderr: stderr.trim(),
error: code !== 0 ? stderr.trim() : null
});
});
child.on('error', (error) => {
clearTimeout(timeout);
resolve({
success: false,
error: error.message
});
});
});
}
/**
* Get the latest commit ID
*/
async _getLatestCommit() {
try {
const result = await this._runGitCommand(['rev-parse', 'HEAD'], 10000);
return result.success ? result.stdout : 'unknown';
} catch (error) {
return 'unknown';
}
}
/**
* Create filesystem backup of important files
*/
async _createFilesystemBackup(backupPath, sessionState) {
try {
const filesDir = path.join(backupPath, 'files');
await fs.ensureDir(filesDir);
// Backup modified files mentioned in session
const modifiedFiles = sessionState.modifiedFiles || sessionState.modified_files || [];
const filesBackedUp = [];
for (const filePath of modifiedFiles) {
try {
const src = path.resolve(filePath);
if (await fs.pathExists(src) && (await fs.stat(src)).isFile()) {
// Create relative path structure
const relativePath = path.relative(this.projectRoot, src);
const dst = path.join(filesDir, relativePath);
await fs.ensureDir(path.dirname(dst));
await fs.copy(src, dst, { preserveTimestamps: true });
filesBackedUp.push(src);
}
} catch (error) {
// Log error but continue with other files
await this._logFileBackupError(filePath, error);
}
}
// Backup important project files
const importantFiles = [
'package.json', 'requirements.txt', 'Cargo.toml',
'pyproject.toml', 'setup.py', '.gitignore',
'README.md', 'CLAUDE.md'
];
for (const fileName of importantFiles) {
const filePath = path.join(this.projectRoot, fileName);
if (await fs.pathExists(filePath)) {
try {
const dst = path.join(filesDir, fileName);
await fs.copy(filePath, dst, { preserveTimestamps: true });
filesBackedUp.push(filePath);
} catch (error) {
// Not critical
}
}
}
return {
success: true,
message: `Backed up ${filesBackedUp.length} files`,
metadata: { files: filesBackedUp }
};
} catch (error) {
return { success: false, error: error.message };
}
}
/**
* Backup session state and context
*/
async _backupSessionState(backupPath, sessionState) {
try {
const stateDir = path.join(backupPath, 'state');
// Save session state
await fs.writeJson(path.join(stateDir, 'session.json'), sessionState, { spaces: 2 });
// Copy hook logs if they exist
const logsSource = path.join(this.projectRoot, '.claude_hooks', 'logs');
if (await fs.pathExists(logsSource)) {
const logsDest = path.join(backupPath, 'logs');
await fs.copy(logsSource, logsDest);
}
// Copy patterns database
const patternsSource = path.join(this.projectRoot, '.claude_hooks', 'patterns');
if (await fs.pathExists(patternsSource)) {
const patternsDest = path.join(stateDir, 'patterns');
await fs.copy(patternsSource, patternsDest);
}
return {
success: true,
message: 'Session state backed up'
};
} catch (error) {
return { success: false, error: error.message };
}
}
/**
* Create minimal backup when full backup fails
*/
async _createMinimalBackup(sessionState) {
try {
// At minimum, save session state to a simple file
const emergencyFile = path.join(this.backupDir, 'emergency_backup.json');
const emergencyData = {
timestamp: new Date().toISOString(),
session_state: sessionState,
type: 'emergency_backup'
};
await fs.writeJson(emergencyFile, emergencyData, { spaces: 2 });
return true;
} catch (error) {
return false;
}
}
/**
* Remove old backups to save space
*/
async _cleanupOldBackups() {
try {
// Get all backup directories
const entries = await fs.readdir(this.backupDir, { withFileTypes: true });
const backupDirs = entries
.filter(entry => entry.isDirectory() && entry.name.startsWith('backup_'))
.map(entry => ({
name: entry.name,
path: path.join(this.backupDir, entry.name)
}));
// Sort by creation time (newest first)
const backupsWithStats = await Promise.all(
backupDirs.map(async (backup) => {
const stats = await fs.stat(backup.path);
return { ...backup, mtime: stats.mtime };
})
);
backupsWithStats.sort((a, b) => b.mtime - a.mtime);
// Remove old backups beyond maxBackups
const oldBackups = backupsWithStats.slice(this.maxBackups);
for (const oldBackup of oldBackups) {
await fs.remove(oldBackup.path);
}
} catch (error) {
// Cleanup failures shouldn't break backup
}
}
/**
* Log backup operation
*/
async _logBackup(backupId, decision, success, error = '') {
try {
const logEntry = {
timestamp: new Date().toISOString(),
backup_id: backupId,
reason: decision.reason,
urgency: decision.urgency,
success,
error
};
// Append to log file
await fs.appendFile(this.logFile, JSON.stringify(logEntry) + '\n');
} catch (error) {
// Logging failures shouldn't break backup
}
}
/**
* Log file backup errors
*/
async _logFileBackupError(filePath, error) {
try {
const errorEntry = {
timestamp: new Date().toISOString(),
type: 'file_backup_error',
file_path: filePath,
error: error.message
};
await fs.appendFile(this.logFile, JSON.stringify(errorEntry) + '\n');
} catch (error) {
// Ignore logging errors
}
}
/**
* List available backups
*/
async listBackups() {
const backups = [];
try {
const entries = await fs.readdir(this.backupDir, { withFileTypes: true });
const backupDirs = entries
.filter(entry => entry.isDirectory() && entry.name.startsWith('backup_'))
.map(entry => path.join(this.backupDir, entry.name));
for (const backupDir of backupDirs) {
const metadataFile = path.join(backupDir, 'metadata.json');
if (await fs.pathExists(metadataFile)) {
try {
const metadata = await fs.readJson(metadataFile);
backups.push(metadata);
} catch (error) {
// Skip corrupted metadata
}
}
}
} catch (error) {
// Return empty list on error
}
return backups.sort((a, b) => {
const timeA = a.timestamp || '';
const timeB = b.timestamp || '';
return timeB.localeCompare(timeA);
});
}
/**
* Get list of files in backup
*/
async _getBackupFiles(filesDir, relativeTo = filesDir) {
const files = [];
try {
const entries = await fs.readdir(filesDir, { withFileTypes: true });
for (const entry of entries) {
const fullPath = path.join(filesDir, entry.name);
const relativePath = path.relative(relativeTo, fullPath);
if (entry.isDirectory()) {
const subFiles = await this._getBackupFiles(fullPath, relativeTo);
files.push(...subFiles);
} else {
files.push(relativePath);
}
}
} catch (error) {
// Return what we have
}
return files;
}
}
module.exports = { BackupManager };

388
lib/backup_manager.py
View file

@ -1,388 +0,0 @@
#!/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
View file

@ -1,202 +0,0 @@
#!/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()

156
lib/context-monitor.js Normal file
View file

@ -0,0 +1,156 @@
/**
* Context Monitor - Estimates context usage and triggers backups
*/
const fs = require('fs-extra');
const path = require('path');
class ContextMonitor {
constructor() {
this.estimatedTokens = 0;
this.promptCount = 0;
this.toolExecutions = 0;
this.sessionStartTime = Date.now();
// Configuration
this.maxContextTokens = 200000; // Conservative estimate
this.backupThreshold = 0.85; // 85% of max context
this.timeThresholdMinutes = 30; // 30 minutes
this.toolThreshold = 25; // 25 tool executions
this.lastBackupTime = Date.now();
this.lastBackupToolCount = 0;
}
/**
* Update context estimates from user prompt
*/
updateFromPrompt(promptData) {
this.promptCount++;
// Estimate tokens from prompt
const prompt = promptData.prompt || '';
const estimatedPromptTokens = this._estimateTokens(prompt);
this.estimatedTokens += estimatedPromptTokens;
// Add context size if provided
if (promptData.context_size) {
this.estimatedTokens += promptData.context_size;
}
}
/**
* Update context estimates from tool usage
*/
updateFromToolUse(toolData) {
this.toolExecutions++;
// Estimate tokens from tool parameters and output
const parameters = JSON.stringify(toolData.parameters || {});
const output = toolData.output || '';
const error = toolData.error || '';
const toolTokens = this._estimateTokens(parameters + output + error);
this.estimatedTokens += toolTokens;
// File operations add more context
if (toolData.tool === 'Read' || toolData.tool === 'Edit') {
this.estimatedTokens += 2000; // Typical file size estimate
} else if (toolData.tool === 'Bash') {
this.estimatedTokens += 500; // Command output estimate
}
}
/**
* Check if backup should be triggered
*/
checkBackupTriggers(hookType, data) {
const decisions = [];
// Context threshold trigger
const contextRatio = this.getContextUsageRatio();
if (contextRatio > this.backupThreshold) {
decisions.push({
shouldBackup: true,
reason: `Context usage ${(contextRatio * 100).toFixed(1)}%`,
urgency: 'high'
});
}
// Time-based trigger
const sessionMinutes = (Date.now() - this.sessionStartTime) / (1000 * 60);
const timeSinceBackup = (Date.now() - this.lastBackupTime) / (1000 * 60);
if (timeSinceBackup > this.timeThresholdMinutes) {
decisions.push({
shouldBackup: true,
reason: `${this.timeThresholdMinutes} minutes since last backup`,
urgency: 'medium'
});
}
// Tool-based trigger
const toolsSinceBackup = this.toolExecutions - this.lastBackupToolCount;
if (toolsSinceBackup >= this.toolThreshold) {
decisions.push({
shouldBackup: true,
reason: `${this.toolThreshold} tools since last backup`,
urgency: 'medium'
});
}
// Return highest priority decision
if (decisions.length > 0) {
const urgencyOrder = { high: 3, medium: 2, low: 1 };
decisions.sort((a, b) => urgencyOrder[b.urgency] - urgencyOrder[a.urgency]);
return decisions[0];
}
return { shouldBackup: false };
}
/**
* Get current context usage ratio (0.0 to 1.0)
*/
getContextUsageRatio() {
return Math.min(1.0, this.estimatedTokens / this.maxContextTokens);
}
/**
* Mark that a backup was performed
*/
markBackupPerformed() {
this.lastBackupTime = Date.now();
this.lastBackupToolCount = this.toolExecutions;
}
/**
* Estimate tokens from text (rough approximation)
*/
_estimateTokens(text) {
if (!text) return 0;
// Rough estimate: ~4 characters per token for English text
// Add some buffer for formatting and special tokens
return Math.ceil(text.length / 3.5);
}
/**
* Get context usage statistics
*/
getStats() {
const sessionMinutes = (Date.now() - this.sessionStartTime) / (1000 * 60);
return {
estimatedTokens: this.estimatedTokens,
contextUsageRatio: this.getContextUsageRatio(),
promptCount: this.promptCount,
toolExecutions: this.toolExecutions,
sessionMinutes: Math.round(sessionMinutes),
lastBackupMinutesAgo: Math.round((Date.now() - this.lastBackupTime) / (1000 * 60))
};
}
}
module.exports = { ContextMonitor };

321
lib/context_monitor.py
View file

@ -1,321 +0,0 @@
#!/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
View file

@ -1,198 +0,0 @@
#!/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

375
lib/session-state.js Normal file
View file

@ -0,0 +1,375 @@
/**
* Session State Manager - Tracks session state and creates continuation docs
*/
const fs = require('fs-extra');
const path = require('path');
class SessionStateManager {
constructor(stateDir = '.claude_hooks') {
this.stateDir = path.resolve(stateDir);
this.sessionId = this._generateSessionId();
this.startTime = Date.now();
// Session data
this.modifiedFiles = new Set();
this.commandsExecuted = [];
this.toolUsage = {};
this.backupHistory = [];
this.contextSnapshots = [];
fs.ensureDirSync(this.stateDir);
this._loadPersistentState();
}
/**
* Update state from tool usage
*/
async updateFromToolUse(toolData) {
const tool = toolData.tool || 'Unknown';
// Track tool usage
this.toolUsage[tool] = (this.toolUsage[tool] || 0) + 1;
// Track file modifications
if (tool === 'Edit' || tool === 'Write' || tool === 'MultiEdit') {
const filePath = toolData.parameters?.file_path;
if (filePath) {
this.modifiedFiles.add(filePath);
}
}
// Track commands
if (tool === 'Bash') {
const command = toolData.parameters?.command;
if (command) {
this.commandsExecuted.push({
command,
timestamp: new Date().toISOString(),
success: toolData.success !== false
});
}
}
// Persist state periodically
if (this.commandsExecuted.length % 5 === 0) {
await this._savePersistentState();
}
}
/**
* Add backup to history
*/
async addBackup(backupId, info) {
this.backupHistory.push({
backupId,
timestamp: new Date().toISOString(),
...info
});
await this._savePersistentState();
}
/**
* Add context snapshot
*/
async addContextSnapshot(snapshot) {
this.contextSnapshots.push({
timestamp: new Date().toISOString(),
...snapshot
});
// Keep only last 10 snapshots
if (this.contextSnapshots.length > 10) {
this.contextSnapshots = this.contextSnapshots.slice(-10);
}
}
/**
* Get comprehensive session summary
*/
async getSessionSummary() {
const duration = Date.now() - this.startTime;
const durationMinutes = Math.round(duration / (1000 * 60));
return {
sessionId: this.sessionId,
startTime: new Date(this.startTime).toISOString(),
duration: duration,
modifiedFiles: Array.from(this.modifiedFiles),
commandsExecuted: this.commandsExecuted,
toolUsage: this.toolUsage,
backupHistory: this.backupHistory,
contextSnapshots: this.contextSnapshots,
sessionStats: {
durationMinutes,
totalToolCalls: Object.values(this.toolUsage).reduce((sum, count) => sum + count, 0),
totalCommands: this.commandsExecuted.length,
filesModified: this.modifiedFiles.size
}
};
}
/**
* Create continuation documentation files
*/
async createContinuationDocs() {
try {
const summary = await this.getSessionSummary();
// Create LAST_SESSION.md
await this._createLastSessionDoc(summary);
// Create/update ACTIVE_TODOS.md (if todos exist)
await this._updateActiveTodos();
} catch (error) {
console.error('Error creating continuation docs:', error.message);
}
}
/**
* Create LAST_SESSION.md with session summary
*/
async _createLastSessionDoc(summary) {
let content = `# Last Claude Session Summary
## Session Overview
- **Session ID**: ${summary.sessionId}
- **Started**: ${summary.startTime}
- **Duration**: ${summary.sessionStats.durationMinutes} minutes
- **Total Tools Used**: ${summary.sessionStats.totalToolCalls}
- **Commands Executed**: ${summary.sessionStats.totalCommands}
- **Files Modified**: ${summary.sessionStats.filesModified}
## Files Modified
`;
if (summary.modifiedFiles.length > 0) {
for (const file of summary.modifiedFiles) {
content += `- ${file}\n`;
}
} else {
content += '*No files were modified in this session*\n';
}
content += `
## Tools Used
`;
for (const [tool, count] of Object.entries(summary.toolUsage)) {
content += `- **${tool}**: ${count} times\n`;
}
content += `
## Recent Commands
`;
const recentCommands = summary.commandsExecuted.slice(-10);
if (recentCommands.length > 0) {
for (const cmd of recentCommands) {
const status = cmd.success ? '✅' : '❌';
const time = new Date(cmd.timestamp).toLocaleTimeString();
content += `- ${status} ${time}: \`${cmd.command}\`\n`;
}
} else {
content += '*No commands executed in this session*\n';
}
if (summary.backupHistory.length > 0) {
content += `
## Backups Created
`;
for (const backup of summary.backupHistory) {
const status = backup.success ? '✅' : '❌';
const time = new Date(backup.timestamp).toLocaleTimeString();
content += `- ${status} ${time}: ${backup.backupId} - ${backup.reason}\n`;
}
}
content += `
## Context Usage Timeline
`;
if (summary.contextSnapshots.length > 0) {
for (const snapshot of summary.contextSnapshots) {
const time = new Date(snapshot.timestamp).toLocaleTimeString();
const usage = ((snapshot.usageRatio || 0) * 100).toFixed(1);
content += `- ${time}: ${usage}% (${snapshot.promptCount || 0} prompts, ${snapshot.toolExecutions || 0} tools)\n`;
}
}
content += `
## Quick Recovery
\`\`\`bash
# Check current project status
git status
# View recent changes
git diff
# List backup directories
ls .claude_hooks/backups/
\`\`\`
*Generated by Claude Hooks on ${new Date().toISOString()}*
`;
await fs.writeFile('LAST_SESSION.md', content);
}
/**
* Update ACTIVE_TODOS.md if todos exist
*/
async _updateActiveTodos() {
// Check if there's an existing ACTIVE_TODOS.md or any todo-related files
const todoFiles = ['ACTIVE_TODOS.md', 'TODO.md', 'todos.md'];
for (const todoFile of todoFiles) {
if (await fs.pathExists(todoFile)) {
// File exists, don't overwrite it
return;
}
}
// Look for todo comments in recently modified files
const todos = await this._extractTodosFromFiles();
if (todos.length > 0) {
let content = `# Active TODOs
*Auto-generated from code comments and session analysis*
`;
for (const todo of todos) {
content += `- [ ] ${todo.text} (${todo.file}:${todo.line})\n`;
}
content += `
*Update this file manually or use Claude to manage your todos*
`;
await fs.writeFile('ACTIVE_TODOS.md', content);
}
}
/**
* Extract TODO comments from modified files
*/
async _extractTodosFromFiles() {
const todos = [];
const todoPattern = /(?:TODO|FIXME|HACK|XXX|NOTE):\s*(.+)/gi;
for (const filePath of this.modifiedFiles) {
try {
if (await fs.pathExists(filePath)) {
const content = await fs.readFile(filePath, 'utf8');
const lines = content.split('\n');
lines.forEach((line, index) => {
const match = todoPattern.exec(line);
if (match) {
todos.push({
text: match[1].trim(),
file: filePath,
line: index + 1
});
}
});
}
} catch (error) {
// Skip files that can't be read
}
}
return todos;
}
/**
* Clean up session resources
*/
async cleanupSession() {
// Save final state
await this._savePersistentState();
// Clean up old session files (keep last 5)
await this._cleanupOldSessions();
}
/**
* Generate unique session ID
*/
_generateSessionId() {
const timestamp = new Date().toISOString()
.replace(/[:-]/g, '')
.replace(/\.\d{3}Z$/, '')
.replace('T', '_');
return `sess_${timestamp}`;
}
/**
* Load persistent state from disk
*/
async _loadPersistentState() {
try {
const stateFile = path.join(this.stateDir, 'session_state.json');
if (await fs.pathExists(stateFile)) {
const state = await fs.readJson(stateFile);
// Only load if session is recent (within 24 hours)
const stateAge = Date.now() - new Date(state.startTime).getTime();
if (stateAge < 24 * 60 * 60 * 1000) {
this.modifiedFiles = new Set(state.modifiedFiles || []);
this.commandsExecuted = state.commandsExecuted || [];
this.toolUsage = state.toolUsage || {};
this.backupHistory = state.backupHistory || [];
this.contextSnapshots = state.contextSnapshots || [];
}
}
} catch (error) {
// Start fresh if loading fails
}
}
/**
* Save persistent state to disk
*/
async _savePersistentState() {
try {
const stateFile = path.join(this.stateDir, 'session_state.json');
const state = {
sessionId: this.sessionId,
startTime: new Date(this.startTime).toISOString(),
modifiedFiles: Array.from(this.modifiedFiles),
commandsExecuted: this.commandsExecuted,
toolUsage: this.toolUsage,
backupHistory: this.backupHistory,
contextSnapshots: this.contextSnapshots,
lastUpdated: new Date().toISOString()
};
await fs.writeJson(stateFile, state, { spaces: 2 });
} catch (error) {
// Don't let save failures break the session
}
}
/**
* Clean up old session state files
*/
async _cleanupOldSessions() {
try {
const statePattern = path.join(this.stateDir, 'session_*.json');
// This is a simple cleanup - in a full implementation,
// you'd use glob patterns to find and clean old files
} catch (error) {
// Ignore cleanup errors
}
}
}
module.exports = { SessionStateManager };

348
lib/session_state.py
View file

@ -1,348 +0,0 @@
#!/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

579
lib/shadow-learner.js Normal file
View file

@ -0,0 +1,579 @@
/**
* Shadow Learner - Pattern learning and prediction system
* Node.js implementation
*/
const fs = require('fs-extra');
const path = require('path');
class ConfidenceCalculator {
/**
* Calculate confidence for command failure patterns
*/
static calculateCommandConfidence(successCount, failureCount, recencyFactor) {
const totalAttempts = successCount + failureCount;
if (totalAttempts === 0) return 0.0;
// Base confidence from failure rate
const failureRate = failureCount / totalAttempts;
// Sample size adjustment (more data = more confidence)
const sampleFactor = Math.min(1.0, totalAttempts / 10.0); // Plateau at 10 samples
// Time decay (recent failures are more relevant)
const confidence = failureRate * sampleFactor * (0.5 + 0.5 * recencyFactor);
return Math.min(0.99, Math.max(0.1, confidence)); // Clamp between 0.1 and 0.99
}
/**
* Calculate confidence for tool sequence patterns
*/
static calculateSequenceConfidence(successfulSequences, totalSequences) {
if (totalSequences === 0) return 0.0;
const successRate = successfulSequences / totalSequences;
const sampleFactor = Math.min(1.0, totalSequences / 5.0);
return successRate * sampleFactor;
}
}
class PatternMatcher {
constructor(db) {
this.db = db;
}
/**
* Find similar command patterns using fuzzy matching
*/
fuzzyCommandMatch(command, threshold = 0.8) {
const cmdTokens = command.toLowerCase().split(' ');
if (cmdTokens.length === 0) return [];
const baseCmd = cmdTokens[0];
const matches = [];
for (const pattern of Object.values(this.db.commandPatterns)) {
const patternCmd = (pattern.trigger.command || '').toLowerCase();
// Exact match
if (patternCmd === baseCmd) {
matches.push(pattern);
}
// Fuzzy match on command name
else if (this._similarity(patternCmd, baseCmd) > threshold) {
matches.push(pattern);
}
// Partial match (e.g., "pip3" matches "pip install")
else if (cmdTokens.some(token => patternCmd.includes(token))) {
matches.push(pattern);
}
}
return matches.sort((a, b) => b.confidence - a.confidence);
}
/**
* Match patterns based on current context
*/
contextPatternMatch(currentContext) {
const matches = [];
for (const pattern of Object.values(this.db.contextPatterns)) {
if (this._contextMatches(currentContext, pattern.trigger)) {
matches.push(pattern);
}
}
return matches.sort((a, b) => b.confidence - a.confidence);
}
/**
* Simple string similarity (Jaccard similarity)
*/
_similarity(str1, str2) {
const set1 = new Set(str1.split(''));
const set2 = new Set(str2.split(''));
const intersection = new Set([...set1].filter(x => set2.has(x)));
const union = new Set([...set1, ...set2]);
return intersection.size / union.size;
}
/**
* Check if current context matches trigger conditions
*/
_contextMatches(current, trigger) {
for (const [key, expectedValue] of Object.entries(trigger)) {
if (!(key in current)) return false;
const currentValue = current[key];
// Handle different value types
if (typeof expectedValue === 'string' && typeof currentValue === 'string') {
if (!currentValue.toLowerCase().includes(expectedValue.toLowerCase())) {
return false;
}
} else if (expectedValue !== currentValue) {
return false;
}
}
return true;
}
}
class LearningEngine {
constructor(db) {
this.db = db;
this.confidenceCalc = ConfidenceCalculator;
}
/**
* Main learning entry point
*/
learnFromExecution(execution) {
// Learn command patterns
if (execution.tool === 'Bash') {
this._learnCommandPattern(execution);
}
// Learn tool sequences
this._learnSequencePattern(execution);
// Learn context patterns
if (!execution.success) {
this._learnFailureContext(execution);
}
}
/**
* Learn from bash command executions
*/
_learnCommandPattern(execution) {
const command = execution.parameters.command || '';
if (!command) return;
const baseCmd = command.split(' ')[0];
const patternId = `cmd_${baseCmd}`;
if (patternId in this.db.commandPatterns) {
const pattern = this.db.commandPatterns[patternId];
// Update statistics
if (execution.success) {
pattern.prediction.successCount = (pattern.prediction.successCount || 0) + 1;
} else {
pattern.prediction.failureCount = (pattern.prediction.failureCount || 0) + 1;
}
// Recalculate confidence
const recency = this._calculateRecency(execution.timestamp);
pattern.confidence = this.confidenceCalc.calculateCommandConfidence(
pattern.prediction.successCount || 0,
pattern.prediction.failureCount || 0,
recency
);
pattern.lastSeen = execution.timestamp;
pattern.evidenceCount += 1;
} else {
// Create new pattern
this.db.commandPatterns[patternId] = {
patternId,
patternType: 'command_execution',
trigger: { command: baseCmd },
prediction: {
successCount: execution.success ? 1 : 0,
failureCount: execution.success ? 0 : 1,
commonErrors: execution.errorMessage ? [execution.errorMessage] : []
},
confidence: 0.3, // Start with low confidence
evidenceCount: 1,
lastSeen: execution.timestamp,
successRate: execution.success ? 1.0 : 0.0
};
}
}
/**
* Learn from tool sequence patterns
*/
_learnSequencePattern(execution) {
// Get recent tool history (last 5 tools)
const recentTools = this.db.executionHistory.slice(-5).map(e => e.tool);
recentTools.push(execution.tool);
// Look for sequences of 2-3 tools
for (let seqLen = 2; seqLen <= 3; seqLen++) {
if (recentTools.length >= seqLen) {
const sequence = recentTools.slice(-seqLen);
const patternId = `seq_${sequence.join('_')}`;
// Update or create sequence pattern
// (Simplified implementation - could be expanded)
}
}
}
/**
* Learn from failure contexts
*/
_learnFailureContext(execution) {
if (!execution.errorMessage) return;
// Extract key error indicators
const errorKey = this._extractErrorKey(execution.errorMessage);
if (!errorKey) return;
const patternId = `ctx_error_${errorKey}`;
if (patternId in this.db.contextPatterns) {
const pattern = this.db.contextPatterns[patternId];
pattern.evidenceCount += 1;
pattern.lastSeen = execution.timestamp;
// Update confidence based on repeated failures
pattern.confidence = Math.min(0.95, pattern.confidence + 0.05);
} else {
// Create new context pattern
this.db.contextPatterns[patternId] = {
patternId,
patternType: 'context_error',
trigger: {
tool: execution.tool,
errorType: errorKey
},
prediction: {
likelyError: execution.errorMessage,
suggestions: this._generateSuggestions(execution)
},
confidence: 0.4,
evidenceCount: 1,
lastSeen: execution.timestamp,
successRate: 0.0
};
}
}
/**
* Calculate recency factor (1.0 = very recent, 0.0 = very old)
*/
_calculateRecency(timestamp) {
const now = new Date();
const ageHours = (now - new Date(timestamp)) / (1000 * 60 * 60);
// Exponential decay: recent events matter more
return Math.max(0.0, Math.exp(-ageHours / 24.0)); // 24 hour half-life
}
/**
* Extract key error indicators from error messages
*/
_extractErrorKey(errorMessage) {
const message = errorMessage.toLowerCase();
const errorPatterns = {
'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 (const [errorType, patterns] of Object.entries(errorPatterns)) {
if (patterns.some(pattern => message.includes(pattern))) {
return errorType;
}
}
return null;
}
/**
* Generate suggestions based on failed execution
*/
_generateSuggestions(execution) {
const suggestions = [];
if (execution.tool === 'Bash') {
const command = execution.parameters.command || '';
if (command) {
const baseCmd = command.split(' ')[0];
// Common command alternatives
const alternatives = {
'pip': ['pip3', 'python -m pip', 'python3 -m pip'],
'python': ['python3'],
'node': ['nodejs'],
'vim': ['nvim', 'nano']
};
if (baseCmd in alternatives) {
const remainingArgs = command.split(' ').slice(1).join(' ');
suggestions.push(
...alternatives[baseCmd].map(alt => `Try '${alt} ${remainingArgs}'`)
);
}
}
}
return suggestions;
}
}
class PredictionEngine {
constructor(matcher) {
this.matcher = matcher;
}
/**
* Predict if a command will succeed and suggest alternatives
*/
predictCommandOutcome(command, context = {}) {
// Find matching patterns
const commandPatterns = this.matcher.fuzzyCommandMatch(command);
const contextPatterns = this.matcher.contextPatternMatch(context);
const prediction = {
likelySuccess: true,
confidence: 0.5,
warnings: [],
suggestions: []
};
// Analyze command patterns
for (const pattern of commandPatterns.slice(0, 3)) { // Top 3 matches
if (pattern.confidence > 0.7) {
const failureRate = (pattern.prediction.failureCount || 0) / Math.max(1, pattern.evidenceCount);
if (failureRate > 0.6) { // High failure rate
prediction.likelySuccess = false;
prediction.confidence = pattern.confidence;
prediction.warnings.push(`Command '${command.split(' ')[0]}' often fails`);
// Add suggestions from pattern
const suggestions = pattern.prediction.suggestions || [];
prediction.suggestions.push(...suggestions);
}
}
}
return prediction;
}
}
class ShadowLearner {
constructor(storagePath = '.claude_hooks/patterns') {
this.storagePath = path.resolve(storagePath);
fs.ensureDirSync(this.storagePath);
this.db = this._loadDatabase();
this.matcher = new PatternMatcher(this.db);
this.learningEngine = new LearningEngine(this.db);
this.predictionEngine = new PredictionEngine(this.matcher);
// Performance cache (simple in-memory cache)
this.predictionCache = new Map();
this.cacheTimeout = 5 * 60 * 1000; // 5 minutes
}
/**
* Learn from tool execution
*/
learnFromExecution(execution) {
try {
this.learningEngine.learnFromExecution(execution);
this.db.executionHistory.push(execution);
// Trim history to keep memory usage reasonable
if (this.db.executionHistory.length > 1000) {
this.db.executionHistory = this.db.executionHistory.slice(-500);
}
} catch (error) {
// Learning failures shouldn't break the system
console.error('Shadow learner error:', error.message);
}
}
/**
* Predict command outcome with caching
*/
predictCommandOutcome(command, context = {}) {
const cacheKey = `cmd_pred:${this._hash(command)}`;
const now = Date.now();
// Check cache
if (this.predictionCache.has(cacheKey)) {
const cached = this.predictionCache.get(cacheKey);
if (now - cached.timestamp < this.cacheTimeout) {
return cached.prediction;
}
}
const prediction = this.predictionEngine.predictCommandOutcome(command, context);
// Cache result
this.predictionCache.set(cacheKey, { prediction, timestamp: now });
// Clean old cache entries
this._cleanCache();
return prediction;
}
/**
* Quick method for command failure learning (backward compatibility)
*/
learnCommandFailure(command, suggestion, confidence) {
const execution = {
tool: 'Bash',
parameters: { command },
success: false,
timestamp: new Date(),
errorMessage: `Command failed: ${command}`
};
this.learnFromExecution(execution);
// Also store the specific suggestion
const baseCmd = command.split(' ')[0];
const patternId = `cmd_${baseCmd}`;
if (patternId in this.db.commandPatterns) {
const pattern = this.db.commandPatterns[patternId];
pattern.prediction.suggestions = pattern.prediction.suggestions || [];
if (!pattern.prediction.suggestions.includes(suggestion)) {
pattern.prediction.suggestions.push(suggestion);
}
}
}
/**
* Get suggestion for a command (backward compatibility)
*/
getSuggestion(command) {
const prediction = this.predictCommandOutcome(command);
if (!prediction.likelySuccess && prediction.suggestions.length > 0) {
return {
suggestion: prediction.suggestions[0].replace(/^Try '|'$/g, ''), // Clean format
confidence: prediction.confidence
};
}
return null;
}
/**
* Save learned patterns to disk
*/
async saveDatabase() {
try {
const patternsFile = path.join(this.storagePath, 'patterns.json');
const backupFile = path.join(this.storagePath, 'patterns.backup.json');
// Create backup of existing data
if (await fs.pathExists(patternsFile)) {
await fs.move(patternsFile, backupFile, { overwrite: true });
}
// Save new data
await fs.writeJson(patternsFile, this._serializeDatabase(), { spaces: 2 });
} catch (error) {
// Save failures shouldn't break the system
console.error('Failed to save shadow learner database:', error.message);
}
}
/**
* Load patterns database from disk
*/
_loadDatabase() {
const patternsFile = path.join(this.storagePath, 'patterns.json');
try {
if (fs.existsSync(patternsFile)) {
const data = fs.readJsonSync(patternsFile);
return this._deserializeDatabase(data);
}
} catch (error) {
// If loading fails, start with empty database
console.error('Failed to load shadow learner database, starting fresh:', error.message);
}
return {
commandPatterns: {},
contextPatterns: {},
sequencePatterns: {},
executionHistory: []
};
}
/**
* Serialize database for JSON storage
*/
_serializeDatabase() {
return {
commandPatterns: this.db.commandPatterns,
contextPatterns: this.db.contextPatterns,
sequencePatterns: this.db.sequencePatterns || {},
executionHistory: this.db.executionHistory.slice(-100), // Keep last 100 executions
metadata: {
version: '1.0.0',
lastSaved: new Date().toISOString()
}
};
}
/**
* Deserialize database from JSON
*/
_deserializeDatabase(data) {
return {
commandPatterns: data.commandPatterns || {},
contextPatterns: data.contextPatterns || {},
sequencePatterns: data.sequencePatterns || {},
executionHistory: (data.executionHistory || []).map(e => ({
...e,
timestamp: new Date(e.timestamp)
}))
};
}
/**
* Simple hash function for cache keys
*/
_hash(str) {
let hash = 0;
for (let i = 0; i < str.length; i++) {
const char = str.charCodeAt(i);
hash = ((hash << 5) - hash) + char;
hash = hash & hash; // Convert to 32bit integer
}
return hash.toString();
}
/**
* Clean expired cache entries
*/
_cleanCache() {
const now = Date.now();
for (const [key, value] of this.predictionCache.entries()) {
if (now - value.timestamp > this.cacheTimeout) {
this.predictionCache.delete(key);
}
}
}
}
module.exports = {
ShadowLearner,
ConfidenceCalculator,
PatternMatcher,
LearningEngine,
PredictionEngine
};

395
lib/shadow_learner.py
View file

@ -1,395 +0,0 @@
#!/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()