#!/usr/bin/env python3 """ Genesis Evolution Protocol ========================== Standardized Protocol for Cross-Session Learning and Evolutionary Development Every agent in the Genesis system MUST follow this protocol to ensure: 1. All learnings are captured and stored 2. Actions and their outcomes are tracked 3. Tasks flow through the memory system 4. Patterns emerge across sessions 5. The system evolves continuously Usage: from genesis_evolution_protocol import EvolutionProtocol, EventType # Initialize at session start protocol = EvolutionProtocol(agent_id="claude_code") protocol.start_session() # Throughout session - capture everything protocol.capture_learning("Discovered that parallel execution improves performance") protocol.capture_action("refactor_memory", expected="cleaner code", actual="30% faster") protocol.capture_task("implement_vectors", status="completed", outcome="3 backends working") protocol.capture_decision("chose_dragonfly", reason="HNSW support", alternatives=["redis", "keydb"]) protocol.capture_error("chromadb_api", error="v1 deprecated", resolution="use v2 endpoint") protocol.capture_pattern("session_handoffs", description="structured HANDOFF.md updates") protocol.capture_capability("vector_search", description="semantic similarity across backends") # End session - consolidate and sync summary = protocol.end_session() Protocol Compliance: All Genesis agents MUST: 1. Call start_session() at initialization 2. Capture ALL significant events during operation 3. Call end_session() before termination 4. Use the standard EventType categories 5. Include context and metadata for each event """ import json import hashlib from datetime import datetime from typing import Dict, List, Any, Optional, Callable from dataclasses import dataclass, field, asdict from enum import Enum from pathlib import Path from functools import wraps import threading import traceback # Import Genesis memory system try: from genesis_memory_cortex import MemoryCortex, MemoryTier CORTEX_AVAILABLE = True except ImportError: CORTEX_AVAILABLE = False MemoryCortex = None class EventType(Enum): """Categories of evolutionary events to capture.""" LEARNING = "learning" # Insights, discoveries, new knowledge ACTION = "action" # Operations performed with outcomes TASK = "task" # Task lifecycle events DECISION = "decision" # Choices made with reasoning ERROR = "error" # Failures and their resolutions PATTERN = "pattern" # Recurring behaviors identified CAPABILITY = "capability" # New abilities gained REFLECTION = "reflection" # Meta-observations about self COLLABORATION = "collaboration" # Multi-agent interactions HANDOFF = "handoff" # Session transition events class EventPriority(Enum): """Priority levels for evolutionary scoring.""" CRITICAL = 1.0 # Must persist - breakthrough insights HIGH = 0.8 # Important - significant learnings MEDIUM = 0.6 # Notable - useful patterns LOW = 0.4 # Minor - contextual details TRACE = 0.2 # Debug - implementation details @dataclass class EvolutionEvent: """A single evolutionary event to be captured.""" id: str event_type: EventType priority: EventPriority content: str timestamp: str agent_id: str session_id: str metadata: Dict[str, Any] = field(default_factory=dict) context: Dict[str, Any] = field(default_factory=dict) relations: List[str] = field(default_factory=list) def to_dict(self) -> Dict: d = asdict(self) d['event_type'] = self.event_type.value d['priority'] = self.priority.value return d def to_memory_content(self) -> str: """Format for memory storage.""" prefix = f"[{self.event_type.value.upper()}]" if self.metadata: meta_str = " | ".join(f"{k}={v}" for k, v in self.metadata.items() if v) return f"{prefix} {self.content} ({meta_str})" return f"{prefix} {self.content}" @dataclass class SessionContext: """Context for the current session.""" session_id: str agent_id: str start_time: str parent_session: Optional[str] = None objectives: List[str] = field(default_factory=list) events: List[EvolutionEvent] = field(default_factory=list) stats: Dict[str, int] = field(default_factory=lambda: { "learnings": 0, "actions": 0, "tasks": 0, "decisions": 0, "errors": 0, "patterns": 0, "capabilities": 0 }) class EvolutionProtocol: """ The Genesis Evolution Protocol. Ensures all agents capture learnings and contribute to system evolution. """ # Persistence paths SESSIONS_DIR = Path("/mnt/e/genesis-system/evolution_sessions") CURRENT_SESSION_FILE = Path("/mnt/e/genesis-system/current_session.json") EVOLUTION_LOG = Path("/mnt/e/genesis-system/evolution_log.jsonl") def __init__(self, agent_id: str, auto_persist: bool = True): """ Initialize the evolution protocol for an agent. Args: agent_id: Unique identifier for this agent (e.g., "claude_code", "claude_desktop") auto_persist: Whether to auto-save events as they occur """ self.agent_id = agent_id self.auto_persist = auto_persist self.session: Optional[SessionContext] = None self._lock = threading.Lock() # Initialize memory cortex if available self.cortex = None if CORTEX_AVAILABLE: try: self.cortex = MemoryCortex(enable_vectors=True) except Exception: pass # Ensure directories exist self.SESSIONS_DIR.mkdir(parents=True, exist_ok=True) def start_session(self, objectives: List[str] = None, parent_session: str = None) -> str: """ Start a new evolution session. MUST be called at agent initialization. Args: objectives: What this session aims to achieve parent_session: ID of previous session (for continuity) Returns: Session ID """ session_id = self._generate_session_id() self.session = SessionContext( session_id=session_id, agent_id=self.agent_id, start_time=datetime.now().isoformat(), parent_session=parent_session, objectives=objectives or [] ) # Load context from parent session if available if parent_session: self._load_parent_context(parent_session) # Record session start self._capture_event( EventType.HANDOFF, f"Session started by {self.agent_id}", priority=EventPriority.MEDIUM, metadata={ "parent_session": parent_session, "objectives_count": len(objectives or []) } ) # Persist current session marker self._save_current_session() return session_id def end_session(self, summary: str = None) -> Dict[str, Any]: """ End the current session. MUST be called before agent termination. Args: summary: Optional summary of session achievements Returns: Session summary with statistics """ if not self.session: return {"error": "No active session"} # Capture session end event self._capture_event( EventType.HANDOFF, summary or f"Session ended by {self.agent_id}", priority=EventPriority.MEDIUM, metadata={ "duration": self._calculate_duration(), "event_count": len(self.session.events) } ) # Consolidate learnings consolidation = self._consolidate_session() # Persist full session self._save_session() # Sync high-value events to memory cortex sync_result = self._sync_to_cortex() result = { "session_id": self.session.session_id, "agent_id": self.agent_id, "duration": self._calculate_duration(), "stats": self.session.stats, "consolidation": consolidation, "sync_result": sync_result } self.session = None return result # ========================================================================== # EVENT CAPTURE METHODS - Use these throughout the session # ========================================================================== def capture_learning(self, content: str, priority: EventPriority = EventPriority.MEDIUM, domain: str = "general", source: str = None, evidence: str = None) -> str: """ Capture a learning or insight. Args: content: What was learned priority: Importance level domain: Knowledge domain (technical, pattern, capability, etc.) source: Where this learning came from evidence: Supporting evidence or example Returns: Event ID """ return self._capture_event( EventType.LEARNING, content, priority=priority, metadata={ "domain": domain, "source": source, "evidence": evidence } ) def capture_action(self, action_name: str, expected: str = None, actual: str = None, success: bool = True, duration_ms: int = None) -> str: """ Capture an action and its outcome. Args: action_name: What was done expected: What was expected to happen actual: What actually happened success: Whether the action succeeded duration_ms: How long it took Returns: Event ID """ # Calculate deviation for learning deviation = 0.0 if expected and actual and expected != actual: deviation = 0.5 # Trigger learning capture priority = EventPriority.LOW if success else EventPriority.HIGH return self._capture_event( EventType.ACTION, f"{action_name}: {actual or expected}", priority=priority, metadata={ "action": action_name, "expected": expected, "actual": actual, "success": success, "deviation": deviation, "duration_ms": duration_ms } ) def capture_task(self, task_name: str, status: str, outcome: str = None, blockers: List[str] = None, next_steps: List[str] = None) -> str: """ Capture a task lifecycle event. Args: task_name: Name of the task status: pending/in_progress/completed/blocked/failed outcome: What resulted from the task blockers: What's blocking progress next_steps: Follow-up actions needed Returns: Event ID """ priority_map = { "completed": EventPriority.MEDIUM, "failed": EventPriority.HIGH, "blocked": EventPriority.HIGH, "in_progress": EventPriority.LOW, "pending": EventPriority.TRACE } return self._capture_event( EventType.TASK, f"{task_name} [{status}]: {outcome or 'in progress'}", priority=priority_map.get(status, EventPriority.MEDIUM), metadata={ "task": task_name, "status": status, "outcome": outcome, "blockers": blockers, "next_steps": next_steps } ) def capture_decision(self, decision: str, reason: str, alternatives: List[str] = None, trade_offs: str = None, reversible: bool = True) -> str: """ Capture a decision with reasoning. Args: decision: What was decided reason: Why this choice was made alternatives: Other options considered trade_offs: What was sacrificed reversible: Can this be undone Returns: Event ID """ priority = EventPriority.HIGH if not reversible else EventPriority.MEDIUM return self._capture_event( EventType.DECISION, f"Decided: {decision} (reason: {reason})", priority=priority, metadata={ "decision": decision, "reason": reason, "alternatives": alternatives, "trade_offs": trade_offs, "reversible": reversible } ) def capture_error(self, context: str, error: str, resolution: str = None, prevention: str = None, stack_trace: str = None) -> str: """ Capture an error and its resolution. Args: context: Where the error occurred error: What went wrong resolution: How it was fixed prevention: How to avoid in future stack_trace: Technical details Returns: Event ID """ priority = EventPriority.HIGH if resolution else EventPriority.CRITICAL return self._capture_event( EventType.ERROR, f"Error in {context}: {error}" + (f" -> Fixed: {resolution}" if resolution else ""), priority=priority, metadata={ "context": context, "error": error, "resolution": resolution, "prevention": prevention, "resolved": resolution is not None }, context={"stack_trace": stack_trace} if stack_trace else {} ) def capture_pattern(self, pattern_name: str, description: str, occurrences: int = 1, examples: List[str] = None, recommendation: str = None) -> str: """ Capture an identified pattern. Args: pattern_name: Name for this pattern description: What the pattern is occurrences: How many times observed examples: Specific instances recommendation: Suggested action Returns: Event ID """ priority = EventPriority.HIGH if occurrences >= 3 else EventPriority.MEDIUM return self._capture_event( EventType.PATTERN, f"Pattern '{pattern_name}': {description}", priority=priority, metadata={ "pattern": pattern_name, "occurrences": occurrences, "examples": examples, "recommendation": recommendation } ) def capture_capability(self, capability: str, description: str, prerequisites: List[str] = None, limitations: List[str] = None, usage_example: str = None) -> str: """ Capture a new capability gained. Args: capability: Name of the capability description: What can now be done prerequisites: What's needed to use it limitations: Known constraints usage_example: How to use it Returns: Event ID """ return self._capture_event( EventType.CAPABILITY, f"New capability: {capability} - {description}", priority=EventPriority.HIGH, metadata={ "capability": capability, "description": description, "prerequisites": prerequisites, "limitations": limitations, "usage_example": usage_example } ) def capture_reflection(self, observation: str, about: str = "self", insight: str = None, action_item: str = None) -> str: """ Capture a meta-observation about the system or self. Args: observation: What was observed about: Subject of reflection (self, system, process) insight: Deeper understanding gained action_item: Suggested improvement Returns: Event ID """ return self._capture_event( EventType.REFLECTION, f"Reflection on {about}: {observation}", priority=EventPriority.MEDIUM, metadata={ "about": about, "insight": insight, "action_item": action_item } ) def capture_collaboration(self, participants: List[str], interaction_type: str, outcome: str, learnings: List[str] = None) -> str: """ Capture a multi-agent collaboration event. Args: participants: Agents involved interaction_type: Type of collaboration (handoff, parallel, sequential) outcome: Result of collaboration learnings: What was learned from working together Returns: Event ID """ return self._capture_event( EventType.COLLABORATION, f"Collaboration ({interaction_type}): {outcome}", priority=EventPriority.MEDIUM, metadata={ "participants": participants, "interaction_type": interaction_type, "outcome": outcome, "learnings": learnings } ) # ========================================================================== # QUERY METHODS - Access past learnings # ========================================================================== def recall_learnings(self, query: str = None, domain: str = None, limit: int = 10) -> List[Dict]: """ Recall past learnings relevant to a query. Args: query: Search query domain: Filter by domain limit: Max results Returns: List of relevant learnings """ if not self.cortex: return self._recall_from_files(query, EventType.LEARNING, limit) results = self.cortex.recall(query or domain or "learning", limit=limit) return [r for r in results if "LEARNING" in r.get("memory", {}).get("content", "")] def recall_patterns(self, limit: int = 10) -> List[Dict]: """Recall identified patterns.""" if not self.cortex: return self._recall_from_files(None, EventType.PATTERN, limit) results = self.cortex.recall("pattern", limit=limit) return [r for r in results if "PATTERN" in r.get("memory", {}).get("content", "")] def recall_capabilities(self) -> List[Dict]: """Recall all known capabilities.""" if not self.cortex: return self._recall_from_files(None, EventType.CAPABILITY, 50) results = self.cortex.recall("capability", limit=50) return [r for r in results if "CAPABILITY" in r.get("memory", {}).get("content", "")] def recall_errors(self, context: str = None, limit: int = 10) -> List[Dict]: """Recall past errors and their resolutions.""" query = f"error {context}" if context else "error resolution" if not self.cortex: return self._recall_from_files(query, EventType.ERROR, limit) results = self.cortex.recall(query, limit=limit) return [r for r in results if "ERROR" in r.get("memory", {}).get("content", "")] # ========================================================================== # PROTOCOL DECORATORS - For automatic capture # ========================================================================== def track_action(self, action_name: str = None): """ Decorator to automatically track function execution as an action. Usage: @protocol.track_action("process_data") def process_data(self, data): return result """ def decorator(func: Callable): @wraps(func) def wrapper(*args, **kwargs): name = action_name or func.__name__ start_time = datetime.now() try: result = func(*args, **kwargs) duration = (datetime.now() - start_time).total_seconds() * 1000 self.capture_action( name, expected="successful execution", actual="completed successfully", success=True, duration_ms=int(duration) ) return result except Exception as e: duration = (datetime.now() - start_time).total_seconds() * 1000 self.capture_error( context=name, error=str(e), stack_trace=traceback.format_exc() ) raise return wrapper return decorator def track_learning(self, domain: str = "general"): """ Decorator to capture return value as a learning. Usage: @protocol.track_learning("optimization") def analyze_performance(self): return "Caching improves latency by 50%" """ def decorator(func: Callable): @wraps(func) def wrapper(*args, **kwargs): result = func(*args, **kwargs) if result and isinstance(result, str): self.capture_learning(result, domain=domain) return result return wrapper return decorator # ========================================================================== # INTERNAL METHODS # ========================================================================== def _capture_event(self, event_type: EventType, content: str, priority: EventPriority = EventPriority.MEDIUM, metadata: Dict = None, context: Dict = None, relations: List[str] = None) -> str: """Internal method to capture any event.""" if not self.session: # Auto-start session if not started self.start_session() event_id = self._generate_event_id(content) event = EvolutionEvent( id=event_id, event_type=event_type, priority=priority, content=content, timestamp=datetime.now().isoformat(), agent_id=self.agent_id, session_id=self.session.session_id, metadata=metadata or {}, context=context or {}, relations=relations or [] ) with self._lock: self.session.events.append(event) # Update stats stat_key = event_type.value + "s" if event_type.value != "capability" else "capabilities" if stat_key in self.session.stats: self.session.stats[stat_key] += 1 # Auto-persist if enabled if self.auto_persist: self._append_to_log(event) # Store in cortex if priority is high enough if priority.value >= EventPriority.MEDIUM.value and self.cortex: try: self.cortex.remember( content=event.to_memory_content(), source=self.agent_id, domain=event_type.value, metadata=event.metadata ) except Exception: pass return event_id def _consolidate_session(self) -> Dict[str, Any]: """Consolidate session learnings into actionable insights.""" if not self.session: return {} # Group events by type by_type = {} for event in self.session.events: key = event.event_type.value if key not in by_type: by_type[key] = [] by_type[key].append(event) # Extract key insights consolidation = { "key_learnings": [], "patterns_identified": [], "capabilities_gained": [], "unresolved_errors": [], "important_decisions": [] } # Top learnings by priority learnings = by_type.get("learning", []) consolidation["key_learnings"] = [ e.content for e in sorted(learnings, key=lambda x: x.priority.value, reverse=True)[:5] ] # Patterns patterns = by_type.get("pattern", []) consolidation["patterns_identified"] = [e.metadata.get("pattern") for e in patterns] # Capabilities capabilities = by_type.get("capability", []) consolidation["capabilities_gained"] = [e.metadata.get("capability") for e in capabilities] # Unresolved errors errors = by_type.get("error", []) consolidation["unresolved_errors"] = [ e.metadata.get("error") for e in errors if not e.metadata.get("resolved") ] # Important decisions decisions = by_type.get("decision", []) consolidation["important_decisions"] = [ {"decision": e.metadata.get("decision"), "reason": e.metadata.get("reason")} for e in decisions if e.priority.value >= EventPriority.HIGH.value ] return consolidation def _sync_to_cortex(self) -> Dict[str, Any]: """Sync high-value events to memory cortex.""" if not self.cortex or not self.session: return {"synced": 0, "reason": "cortex not available"} synced = 0 for event in self.session.events: if event.priority.value >= EventPriority.MEDIUM.value: try: self.cortex.remember( content=event.to_memory_content(), source=self.agent_id, domain=event.event_type.value, metadata=event.metadata ) synced += 1 except Exception: pass return {"synced": synced, "total": len(self.session.events)} def _save_session(self): """Persist full session to file.""" if not self.session: return session_file = self.SESSIONS_DIR / f"{self.session.session_id}.json" data = { "session_id": self.session.session_id, "agent_id": self.session.agent_id, "start_time": self.session.start_time, "end_time": datetime.now().isoformat(), "parent_session": self.session.parent_session, "objectives": self.session.objectives, "stats": self.session.stats, "events": [e.to_dict() for e in self.session.events] } with open(session_file, 'w') as f: json.dump(data, f, indent=2) def _save_current_session(self): """Save current session marker.""" if not self.session: return data = { "session_id": self.session.session_id, "agent_id": self.agent_id, "start_time": self.session.start_time } with open(self.CURRENT_SESSION_FILE, 'w') as f: json.dump(data, f) def _append_to_log(self, event: EvolutionEvent): """Append event to rolling log file.""" with open(self.EVOLUTION_LOG, 'a') as f: f.write(json.dumps(event.to_dict()) + "\n") def _load_parent_context(self, parent_session: str): """Load context from parent session.""" session_file = self.SESSIONS_DIR / f"{parent_session}.json" if not session_file.exists(): return try: with open(session_file) as f: parent_data = json.load(f) # Load unresolved errors and incomplete tasks as context parent_events = parent_data.get("events", []) for event in parent_events: if event.get("event_type") == "error" and not event.get("metadata", {}).get("resolved"): self.session.context["unresolved_errors"] = self.session.context.get("unresolved_errors", []) self.session.context["unresolved_errors"].append(event) if event.get("event_type") == "task" and event.get("metadata", {}).get("status") in ["pending", "in_progress"]: self.session.context["pending_tasks"] = self.session.context.get("pending_tasks", []) self.session.context["pending_tasks"].append(event) except Exception: pass def _recall_from_files(self, query: str, event_type: EventType, limit: int) -> List[Dict]: """Recall from file-based storage when cortex unavailable.""" results = [] if self.EVOLUTION_LOG.exists(): with open(self.EVOLUTION_LOG) as f: for line in f: try: event = json.loads(line) if event.get("event_type") == event_type.value: if not query or query.lower() in event.get("content", "").lower(): results.append(event) except Exception: continue return results[-limit:] def _calculate_duration(self) -> str: """Calculate session duration.""" if not self.session: return "unknown" start = datetime.fromisoformat(self.session.start_time) duration = datetime.now() - start hours, remainder = divmod(duration.seconds, 3600) minutes, seconds = divmod(remainder, 60) if hours > 0: return f"{hours}h {minutes}m" elif minutes > 0: return f"{minutes}m {seconds}s" else: return f"{seconds}s" def _generate_session_id(self) -> str: """Generate unique session ID.""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") suffix = hashlib.md5(f"{self.agent_id}{timestamp}".encode()).hexdigest()[:6] return f"{self.agent_id}_{timestamp}_{suffix}" def _generate_event_id(self, content: str) -> str: """Generate unique event ID.""" timestamp = datetime.now().isoformat() return hashlib.sha256(f"{content}{timestamp}".encode()).hexdigest()[:12] def get_session_stats(self) -> Dict[str, Any]: """Get current session statistics.""" if not self.session: return {"error": "No active session"} return { "session_id": self.session.session_id, "agent_id": self.agent_id, "duration": self._calculate_duration(), "event_count": len(self.session.events), "stats": self.session.stats, "objectives": self.session.objectives } # ========================================================================== # GLOBAL PROTOCOL INSTANCE # ========================================================================== # Singleton protocol instance for easy access _global_protocol: Optional[EvolutionProtocol] = None def get_protocol(agent_id: str = "genesis") -> EvolutionProtocol: """Get or create the global protocol instance.""" global _global_protocol if _global_protocol is None: _global_protocol = EvolutionProtocol(agent_id) return _global_protocol def init_protocol(agent_id: str, objectives: List[str] = None) -> EvolutionProtocol: """Initialize a new protocol session.""" global _global_protocol _global_protocol = EvolutionProtocol(agent_id) _global_protocol.start_session(objectives=objectives) return _global_protocol # ========================================================================== # CLI INTERFACE # ========================================================================== if __name__ == "__main__": import sys print(""" Genesis Evolution Protocol ========================== Protocol for cross-session learning and evolutionary development. Usage in agents: from genesis_evolution_protocol import init_protocol # At start protocol = init_protocol("my_agent", objectives=["task1", "task2"]) # During operation protocol.capture_learning("Discovered X improves Y") protocol.capture_action("did_thing", expected="A", actual="B") protocol.capture_error("context", "error", resolution="fixed by...") # At end summary = protocol.end_session() Commands: python genesis_evolution_protocol.py demo Run a demo session python genesis_evolution_protocol.py stats Show evolution statistics python genesis_evolution_protocol.py recall Recall past learnings """) if len(sys.argv) > 1: command = sys.argv[1] if command == "demo": print("\n=== Running Demo Session ===\n") protocol = EvolutionProtocol("demo_agent") protocol.start_session(objectives=["Demonstrate protocol", "Show all event types"]) # Demonstrate each event type protocol.capture_learning( "Evolution protocol enables cross-session continuity", domain="capability", evidence="This demo session" ) protocol.capture_action( "initialize_protocol", expected="Protocol ready", actual="Protocol ready", success=True ) protocol.capture_task( "demonstrate_events", status="completed", outcome="All event types captured" ) protocol.capture_decision( "use_json_storage", reason="Simple, portable, human-readable", alternatives=["sqlite", "msgpack"] ) protocol.capture_error( "test_error", error="Simulated error for demo", resolution="No fix needed - just a demo" ) protocol.capture_pattern( "session_structure", description="Start -> Events -> End pattern", occurrences=1 ) protocol.capture_capability( "evolution_tracking", description="Track all session events for evolutionary learning" ) summary = protocol.end_session("Demo session completed successfully") print("Session Summary:") print(json.dumps(summary, indent=2)) elif command == "stats": print("\n=== Evolution Statistics ===\n") sessions_dir = Path("/mnt/e/genesis-system/evolution_sessions") if sessions_dir.exists(): sessions = list(sessions_dir.glob("*.json")) print(f"Total sessions: {len(sessions)}") for session_file in sessions[-5:]: with open(session_file) as f: data = json.load(f) print(f"\n{data['session_id']}:") print(f" Agent: {data['agent_id']}") print(f" Events: {len(data.get('events', []))}") print(f" Stats: {data.get('stats', {})}") else: print("No sessions recorded yet.") elif command == "recall": print("\n=== Recalling Past Learnings ===\n") log_file = Path("/mnt/e/genesis-system/evolution_log.jsonl") if log_file.exists(): learnings = [] with open(log_file) as f: for line in f: try: event = json.loads(line) if event.get("event_type") == "learning": learnings.append(event) except: continue print(f"Total learnings: {len(learnings)}") for learning in learnings[-10:]: print(f"\n- {learning['content']}") print(f" [{learning.get('metadata', {}).get('domain', 'general')}] {learning['timestamp']}") else: print("No learnings recorded yet.")