""" AIVA Claude Code Bridge - PM-028 AIVA invokes Claude Code for code tasks. Manages session spawning, context passing, and result collection. """ import os import json import logging import subprocess import tempfile from datetime import datetime from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict, field from pathlib import Path # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @dataclass class ClaudeCodeSession: """Represents a Claude Code session.""" session_id: str task_description: str context: Dict started_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) completed_at: Optional[str] = None status: str = "pending" result: Optional[Any] = None learnings: List[str] = field(default_factory=list) files_modified: List[str] = field(default_factory=list) tokens_used: int = 0 cost_estimate: float = 0.0 @dataclass class ClaudeCodeResult: """Result from a Claude Code invocation.""" session_id: str success: bool output: Optional[str] = None error: Optional[str] = None files_modified: List[str] = field(default_factory=list) learnings: List[str] = field(default_factory=list) execution_time_seconds: float = 0.0 tokens_used: int = 0 class ClaudeCodeBridge: """ Bridge for AIVA to invoke Claude Code sessions. Usage: bridge = ClaudeCodeBridge(memory_bridge) result = bridge.invoke_claude_code( task="Fix the bug in file X", context={"file": "path/to/file.py"}, working_dir="/path/to/project" ) if result.success: print(f"Files modified: {result.files_modified}") """ def __init__(self, memory_bridge=None, log_dir: str = "logs"): """ Initialize the Claude Code bridge. Args: memory_bridge: MemoryBridge for storing learnings log_dir: Directory for session logs """ self.memory_bridge = memory_bridge self.log_dir = Path(log_dir) self.log_dir.mkdir(parents=True, exist_ok=True) self.sessions: Dict[str, ClaudeCodeSession] = {} self.session_log_file = self.log_dir / "claude_code_sessions.jsonl" logger.info("ClaudeCodeBridge initialized") def invoke_claude_code( self, task: str, context: Optional[Dict] = None, working_dir: Optional[str] = None, files: Optional[List[str]] = None, timeout: int = 300, learnings: Optional[List[str]] = None ) -> ClaudeCodeResult: """ Invoke Claude Code for a task. Args: task: Task description for Claude Code context: Additional context to pass working_dir: Working directory for the session files: Specific files to focus on timeout: Timeout in seconds learnings: Previous learnings to pass Returns: ClaudeCodeResult with outcome """ import time start_time = time.time() # Generate session ID session_id = f"cc_{int(time.time() * 1000)}" context = context or {} # Create session record session = ClaudeCodeSession( session_id=session_id, task_description=task, context=context, status="running" ) self.sessions[session_id] = session logger.info(f"Starting Claude Code session {session_id}: {task[:50]}...") try: # Build the prompt prompt = self._build_prompt(task, context, files, learnings) # Execute Claude Code result = self._execute_claude_code( prompt=prompt, working_dir=working_dir or os.getcwd(), timeout=timeout ) execution_time = time.time() - start_time if result["success"]: session.status = "completed" session.result = result.get("output") session.files_modified = result.get("files_modified", []) session.learnings = self._extract_learnings(result.get("output", "")) cc_result = ClaudeCodeResult( session_id=session_id, success=True, output=result.get("output"), files_modified=session.files_modified, learnings=session.learnings, execution_time_seconds=execution_time ) else: session.status = "failed" cc_result = ClaudeCodeResult( session_id=session_id, success=False, error=result.get("error"), execution_time_seconds=execution_time ) session.completed_at = datetime.utcnow().isoformat() # Update memory with learnings if self.memory_bridge and session.learnings: self._store_learnings(session) # Log the session self._log_session(session) logger.info(f"Session {session_id} completed: success={cc_result.success}") return cc_result except Exception as e: execution_time = time.time() - start_time session.status = "error" session.completed_at = datetime.utcnow().isoformat() self._log_session(session) return ClaudeCodeResult( session_id=session_id, success=False, error=str(e), execution_time_seconds=execution_time ) def _build_prompt( self, task: str, context: Dict, files: Optional[List[str]], learnings: Optional[List[str]] ) -> str: """Build the prompt for Claude Code.""" parts = [f"Task: {task}"] if context: parts.append(f"\nContext:\n{json.dumps(context, indent=2)}") if files: parts.append(f"\nFocus on files: {', '.join(files)}") if learnings: parts.append("\nLearnings from previous attempts:") for learning in learnings: parts.append(f" - {learning}") parts.append("\n\nPlease complete this task efficiently and report what you modified.") return "\n".join(parts) def _execute_claude_code( self, prompt: str, working_dir: str, timeout: int ) -> Dict: """ Execute Claude Code subprocess. This is a simulated implementation. In production, this would spawn an actual Claude Code session. """ # Check if claude command is available try: # Try to execute claude command result = subprocess.run( ["claude", "--version"], capture_output=True, text=True, timeout=10 ) claude_available = result.returncode == 0 except (FileNotFoundError, subprocess.TimeoutExpired): claude_available = False if claude_available: # Create temp file with prompt with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f: f.write(prompt) prompt_file = f.name try: result = subprocess.run( ["claude", "--prompt", prompt_file, "--cwd", working_dir], capture_output=True, text=True, timeout=timeout ) return { "success": result.returncode == 0, "output": result.stdout, "error": result.stderr if result.returncode != 0 else None, "files_modified": self._parse_modified_files(result.stdout) } finally: os.unlink(prompt_file) else: # Simulated response for development/testing logger.warning("Claude Code CLI not available, simulating response") return { "success": True, "output": f"[Simulated] Task would be executed: {prompt[:100]}...", "files_modified": [], "simulated": True } def _parse_modified_files(self, output: str) -> List[str]: """Parse output to find modified files.""" files = [] # Look for common patterns in Claude Code output import re patterns = [ r'Modified: (.+)', r'Created: (.+)', r'Updated: (.+)', r'Wrote to: (.+)', ] for pattern in patterns: matches = re.findall(pattern, output) files.extend(matches) return list(set(files)) def _extract_learnings(self, output: str) -> List[str]: """Extract learnings from session output.""" learnings = [] # Look for learning indicators import re patterns = [ r'Learned: (.+)', r'Note: (.+)', r'Important: (.+)', r'Discovered: (.+)', ] for pattern in patterns: matches = re.findall(pattern, output, re.IGNORECASE) learnings.extend(matches) return learnings def _store_learnings(self, session: ClaudeCodeSession) -> None: """Store session learnings in memory.""" if not self.memory_bridge: return try: self.memory_bridge.store_memory( content={ "session_id": session.session_id, "task": session.task_description, "learnings": session.learnings, "files_modified": session.files_modified }, memory_type="entity", metadata={ "entity_type": "claude_code_session", "name": f"Session {session.session_id}" } ) logger.debug(f"Stored learnings for session {session.session_id}") except Exception as e: logger.error(f"Failed to store learnings: {e}") def _log_session(self, session: ClaudeCodeSession) -> None: """Log session to file.""" try: with open(self.session_log_file, "a") as f: f.write(json.dumps(asdict(session)) + "\n") except Exception as e: logger.error(f"Failed to log session: {e}") def get_session(self, session_id: str) -> Optional[ClaudeCodeSession]: """Get a session by ID.""" return self.sessions.get(session_id) def get_recent_learnings(self, limit: int = 10) -> List[str]: """Get recent learnings from all sessions.""" learnings = [] for session in sorted( self.sessions.values(), key=lambda s: s.started_at, reverse=True )[:limit]: learnings.extend(session.learnings) return learnings def get_stats(self) -> Dict: """Get bridge statistics.""" total = len(self.sessions) completed = sum(1 for s in self.sessions.values() if s.status == "completed") failed = sum(1 for s in self.sessions.values() if s.status in ["failed", "error"]) return { "total_sessions": total, "completed": completed, "failed": failed, "success_rate": completed / total if total > 0 else 0.0, "total_learnings": sum(len(s.learnings) for s in self.sessions.values()) } # Singleton instance _bridge: Optional[ClaudeCodeBridge] = None def get_claude_code_bridge(memory_bridge=None) -> ClaudeCodeBridge: """Get or create singleton ClaudeCodeBridge.""" global _bridge if _bridge is None: _bridge = ClaudeCodeBridge(memory_bridge) return _bridge if __name__ == "__main__": # Example usage bridge = ClaudeCodeBridge() # Invoke Claude Code result = bridge.invoke_claude_code( task="Create a simple hello world function", context={"language": "python"}, working_dir=os.getcwd() ) print(f"\nSession result:") print(f" Success: {result.success}") print(f" Output: {result.output[:200] if result.output else 'None'}...") print(f" Files modified: {result.files_modified}") print(f" Learnings: {result.learnings}") print(f" Execution time: {result.execution_time_seconds:.1f}s") # Get stats print(f"\nBridge stats: {bridge.get_stats()}")