archived-selective/cli-backup/context.py

#!/usr/bin/env python3
"""
Stack 2.9 - Context Management Module
Handles project awareness, session memory, and long-term memory integration.
"""

import os
import json
import re
from pathlib import Path
from typing import Any, Dict, List, Optional, Set
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from collections import defaultdict


@dataclass
class ProjectContext:
    """Represents a project's context."""
    name: str
    path: str
    language: Optional[str] = None
    framework: Optional[str] = None
    files: List[str] = field(default_factory=list)
    dirs: List[str] = field(default_factory=list)
    has_git: bool = False
    dependencies: List[str] = field(default_factory=list)
    entry_points: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class SessionMemory:
    """Represents the current session's memory."""
    messages: List[Dict[str, Any]] = field(default_factory=list)
    tools_used: List[str] = field(default_factory=list)
    files_touched: List[str] = field(default_factory=list)
    commands_run: List[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    last_updated: datetime = field(default_factory=datetime.now)
    
    def add_message(self, role: str, content: str, metadata: Optional[Dict] = None):
        """Add a message to session memory."""
        self.messages.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat(),
            "metadata": metadata or {}
        })
        self.last_updated = datetime.now()
    
    def add_tool_usage(self, tool_name: str, result: Any):
        """Record tool usage."""
        self.tools_used.append({
            "tool": tool_name,
            "timestamp": datetime.now().isoformat(),
            "success": result.get("success", False) if isinstance(result, dict) else True
        })
        self.last_updated = datetime.now()
    
    def add_file_touched(self, file_path: str, action: str):
        """Record file access."""
        self.files_touched.append({
            "path": file_path,
            "action": action,
            "timestamp": datetime.now().isoformat()
        })
        self.last_updated = datetime.now()
    
    def add_command(self, command: str, result: Optional[Dict] = None):
        """Record command execution."""
        self.commands_run.append({
            "command": command,
            "result": result,
            "timestamp": datetime.now().isoformat()
        })
        self.last_updated = datetime.now()
    
    def get_summary(self) -> Dict[str, Any]:
        """Get session summary."""
        return {
            "messages_count": len(self.messages),
            "tools_used_count": len(self.tools_used),
            "files_touched_count": len(self.files_touched),
            "commands_run_count": len(self.commands_run),
            "duration_minutes": (datetime.now() - self.created_at).total_seconds() / 60,
            "created_at": self.created_at.isoformat(),
            "last_updated": self.last_updated.isoformat()
        }


class ContextManager:
    """Manages context across projects, sessions, and long-term memory."""
    
    def __init__(self, workspace_path: str = "/Users/walidsobhi/.openclaw/workspace"):
        self.workspace = Path(workspace_path)
        self.session = SessionMemory()
        self.projects: Dict[str, ProjectContext] = {}
        self.current_project: Optional[ProjectContext] = None
        self._load_context()
    
    def _load_context(self):
        """Load existing context files."""
        # Load workspace context files
        context_files = {
            "AGENTS.md": "agents",
            "SOUL.md": "soul",
            "TOOLS.md": "tools",
            "USER.md": "user",
            "MEMORY.md": "memory"
        }
        
        self.context = {}
        for filename, key in context_files.items():
            file_path = self.workspace / filename
            if file_path.exists():
                self.context[key] = file_path.read_text(encoding='utf-8')
        
        # Scan for projects
        self._scan_projects()
    
    def _scan_projects(self):
        """Scan workspace for projects."""
        for item in self.workspace.iterdir():
            if item.is_dir() and not item.name.startswith('.'):
                # Check if it's a project
                if (item / "pyproject.toml").exists() or (item / "package.json").exists():
                    self.projects[item.name] = ProjectContext(
                        name=item.name,
                        path=str(item)
                    )
    
    def load_project(self, project_name: str) -> Optional[ProjectContext]:
        """Load a specific project."""
        project_path = self.workspace / project_name
        
        if not project_path.exists():
            return None
        
        # Create project context
        ctx = ProjectContext(
            name=project_name,
            path=str(project_path)
        )
        
        # Detect language/framework
        if (project_path / "pyproject.toml").exists():
            ctx.language = "python"
            try:
                content = (project_path / "pyproject.toml").read_text()
                if "fastapi" in content:
                    ctx.framework = "fastapi"
                elif "django" in content:
                    ctx.framework = "django"
                elif "flask" in content:
                    ctx.framework = "flask"
            except:
                pass
        
        if (project_path / "package.json").exists():
            ctx.language = "javascript"
            try:
                content = json.loads((project_path / "package.json").read_text())
                ctx.dependencies = list(content.get("dependencies", {}).keys())
                if "next" in content.get("dependencies", {}):
                    ctx.framework = "next"
                elif "react" in content.get("dependencies", {}):
                    ctx.framework = "react"
            except:
                pass
        
        # Check for git
        ctx.has_git = (project_path / ".git").exists()
        
        # Scan files
        try:
            for item in project_path.rglob("*"):
                if len(ctx.files) > 100:
                    break
                rel = item.relative_to(project_path)
                if item.is_file():
                    ctx.files.append(str(rel))
                elif item.is_dir() and not item.name.startswith('.'):
                    ctx.dirs.append(str(rel))
        except:
            pass
        
        # Find entry points
        entry_patterns = ["main.py", "app.py", "index.js", "main.js", "server.py"]
        for pattern in entry_patterns:
            for f in ctx.files:
                if f.endswith(pattern):
                    ctx.entry_points.append(f)
        
        self.projects[project_name] = ctx
        self.current_project = ctx
        return ctx
    
    def get_context_summary(self) -> Dict[str, Any]:
        """Get context summary."""
        return {
            "workspace": str(self.workspace),
            "projects": list(self.projects.keys()),
            "current_project": self.current_project.name if self.current_project else None,
            "session": self.session.get_summary(),
            "has_agents": "agents" in self.context,
            "has_soul": "soul" in self.context,
            "has_tools": "tools" in self.context,
            "has_memory": "memory" in self.context
        }
    
    def get_workspace_context(self) -> str:
        """Get formatted workspace context."""
        lines = ["# Workspace Context"]
        lines.append(f"\n## Projects ({len(self.projects)})")
        
        for name, proj in self.projects.items():
            lines.append(f"- **{name}** ({proj.language or 'unknown'})")
            if proj.framework:
                lines.append(f"  - Framework: {proj.framework}")
            lines.append(f"  - Path: {proj.path}")
            if proj.has_git:
                lines.append("  - Git: ✓")
        
        if self.current_project:
            lines.append(f"\n## Current Project: {self.current_project.name}")
            lines.append(f"- Files: {len(self.current_project.files)}")
            lines.append(f"- Dirs: {len(self.current_project.dirs)}")
            if self.current_project.entry_points:
                lines.append(f"- Entry: {self.current_project.entry_points[0]}")
        
        lines.append(f"\n## Session")
        summary = self.session.get_summary()
        lines.append(f"- Messages: {summary['messages_count']}")
        lines.append(f"- Tools used: {summary['tools_used_count']}")
        lines.append(f"- Files touched: {summary['files_touched_count']}")
        
        return "\n".join(lines)
    
    def search_memory(self, query: str, max_results: int = 5) -> List[Dict[str, Any]]:
        """Search long-term memory."""
        results = []
        
        # Search MEMORY.md
        memory_file = self.workspace / "MEMORY.md"
        if memory_file.exists():
            content = memory_file.read_text()
            if query.lower() in content.lower():
                results.append({
                    "file": str(memory_file),
                    "type": "memory",
                    "content": content[:500]
                })
        
        # Search memory folder
        memory_dir = self.workspace / "memory"
        if memory_dir.exists():
            for f in memory_dir.rglob("*.md"):
                try:
                    content = f.read_text()
                    if query.lower() in content.lower():
                        results.append({
                            "file": str(f),
                            "type": "daily",
                            "content": content[:500]
                        })
                except:
                    continue
        
        return results[:max_results]
    
    def save_to_memory(self, key: str, value: str):
        """Save to long-term memory."""
        memory_file = self.workspace / "MEMORY.md"
        
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
        entry = f"\n### {key}\n_{timestamp}_\n{value}\n"
        
        with open(memory_file, "a") as f:
            f.write(entry)
    
    def get_recent_context(self, days: int = 7) -> List[Dict[str, Any]]:
        """Get recent context from memory."""
        results = []
        
        memory_dir = self.workspace / "memory"
        if memory_dir.exists():
            # Get files from last N days
            cutoff = datetime.now() - timedelta(days=days)
            
            for f in sorted(memory_dir.glob("*.md"), key=lambda x: x.stat().st_mtime, reverse=True):
                try:
                    mtime = datetime.fromtimestamp(f.stat().st_mtime)
                    if mtime > cutoff:
                        content = f.read_text()
                        results.append({
                            "file": str(f),
                            "date": mtime.isoformat(),
                            "content": content[:1000]
                        })
                except:
                    continue
        
        return results


class ProjectAware:
    """Mixin for project-aware functionality."""
    
    def __init__(self):
        self.context_manager = ContextManager()
    
    def detect_project(self, path: str) -> Optional[str]:
        """Detect project from path."""
        p = Path(path).resolve()
        
        # Walk up to find project root
        while p != p.parent:
            for name in ["pyproject.toml", "package.json", "Cargo.toml", "go.mod"]:
                if (p / name).exists():
                    return p.name
            p = p.parent
        
        return None
    
    def get_project_context(self, project_name: str) -> Optional[ProjectContext]:
        """Get project context."""
        return self.context_manager.load_project(project_name)
    
    def format_context_for_prompt(self) -> str:
        """Format context for LLM prompt."""
        return self.context_manager.get_workspace_context()


def create_context_manager(workspace: Optional[str] = None) -> ContextManager:
    """Factory function to create context manager."""
    return ContextManager(workspace or "/Users/walidsobhi/.openclaw/workspace")


if __name__ == "__main__":
    print("Stack 2.9 Context Module")
    cm = ContextManager()
    print(cm.get_context_summary())