#!/usr/bin/env python3 """ Capability Registry - Story 1 ============================= PostgreSQL-based storage for discovered Claude Code capabilities. Uses Elestio Core infrastructure (RULE 6: No SQLite). Tracks: - Discovery metadata (who, when, how) - Specification (usage, parameters, examples) - Implementation status - Verification results - Scoring """ import sys import json from pathlib import Path from datetime import datetime from typing import Dict, List, Any, Optional from dataclasses import dataclass, field from enum import Enum # Elestio PostgreSQL config sys.path.append('/mnt/e/genesis-system/data/genesis-memory') from elestio_config import PostgresConfig import psycopg2 from psycopg2.extras import RealDictCursor class CapabilityStatus(Enum): DISCOVERED = "discovered" IMPLEMENTING = "implementing" IMPLEMENTED = "implemented" VERIFIED = "verified" FAILED = "failed" class CapabilityCategory(Enum): CLI_FEATURES = "cli_features" TOOL_SYSTEM = "tool_system" MCP_INTEGRATION = "mcp_integration" HOOKS_SYSTEM = "hooks_system" SKILLS_SYSTEM = "skills_system" MEMORY_CONTEXT = "memory_context" IDE_INTEGRATION = "ide_integration" ADVANCED_FEATURES = "advanced_features" SECURITY_PERMISSIONS = "security_permissions" EXTENSION_POINTS = "extension_points" @dataclass class Capability: """A discovered Claude Code capability.""" id: str name: str category: str description: str # Discovery discovered_by: str = "" discovered_at: str = "" discovery_source: str = "" discovery_confidence: float = 0.0 # Specification usage: str = "" parameters: List[str] = field(default_factory=list) limitations: List[str] = field(default_factory=list) examples: List[str] = field(default_factory=list) documentation_url: str = "" # Implementation status: str = "discovered" genesis_integration: str = "" implemented_by: str = "" implemented_at: str = "" tests_passing: bool = False # Verification verified_by: str = "" verified_at: str = "" issues_found: List[str] = field(default_factory=list) quality_score: float = 0.0 # Scoring discovery_points: int = 0 implementation_points: int = 0 documentation_points: int = 0 integration_points: int = 0 penalties: int = 0 bonuses: int = 0 @property def total_points(self) -> int: return (self.discovery_points + self.implementation_points + self.documentation_points + self.integration_points + self.bonuses - self.penalties) class CapabilityRegistry: """ PostgreSQL-based registry for Claude Code capabilities. Uses Elestio Core infrastructure. Usage: registry = CapabilityRegistry() # Add a capability cap = Capability( id="cap_001", name="Parallel Tool Execution", category="advanced_features", description="Execute multiple tools in parallel" ) registry.add_capability(cap) # Query capabilities all_caps = registry.get_all() implemented = registry.get_by_status("implemented") """ def __init__(self): self.conn_params = PostgresConfig.get_connection_params() self._init_db() def _get_conn(self): """Get a PostgreSQL connection.""" return psycopg2.connect(**self.conn_params) def _init_db(self): """Initialize database schema.""" with self._get_conn() as conn: with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS capability_registry ( id TEXT PRIMARY KEY, name TEXT NOT NULL, category TEXT NOT NULL, description TEXT, discovered_by TEXT, discovered_at TEXT, discovery_source TEXT, discovery_confidence REAL DEFAULT 0, usage TEXT, parameters JSONB DEFAULT '[]', limitations JSONB DEFAULT '[]', examples JSONB DEFAULT '[]', documentation_url TEXT, status TEXT DEFAULT 'discovered', genesis_integration TEXT, implemented_by TEXT, implemented_at TEXT, tests_passing BOOLEAN DEFAULT FALSE, verified_by TEXT, verified_at TEXT, issues_found JSONB DEFAULT '[]', quality_score REAL DEFAULT 0, discovery_points INTEGER DEFAULT 0, implementation_points INTEGER DEFAULT 0, documentation_points INTEGER DEFAULT 0, integration_points INTEGER DEFAULT 0, penalties INTEGER DEFAULT 0, bonuses INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) cur.execute(""" CREATE TABLE IF NOT EXISTS capability_history ( id SERIAL PRIMARY KEY, capability_id TEXT REFERENCES capability_registry(id), action TEXT, agent TEXT, details TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) cur.execute(""" CREATE INDEX IF NOT EXISTS idx_cap_status ON capability_registry(status) """) cur.execute(""" CREATE INDEX IF NOT EXISTS idx_cap_category ON capability_registry(category) """) conn.commit() def add_capability(self, cap: Capability) -> bool: """Add a new capability to the registry.""" try: with self._get_conn() as conn: with conn.cursor() as cur: cur.execute(""" INSERT INTO capability_registry ( id, name, category, description, discovered_by, discovered_at, discovery_source, discovery_confidence, usage, parameters, limitations, examples, documentation_url, status, genesis_integration, implemented_by, implemented_at, tests_passing, verified_by, verified_at, issues_found, quality_score, discovery_points, implementation_points, documentation_points, integration_points, penalties, bonuses ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( cap.id, cap.name, cap.category, cap.description, cap.discovered_by, cap.discovered_at, cap.discovery_source, cap.discovery_confidence, cap.usage, json.dumps(cap.parameters), json.dumps(cap.limitations), json.dumps(cap.examples), cap.documentation_url, cap.status, cap.genesis_integration, cap.implemented_by, cap.implemented_at, cap.tests_passing, cap.verified_by, cap.verified_at, json.dumps(cap.issues_found), cap.quality_score, cap.discovery_points, cap.implementation_points, cap.documentation_points, cap.integration_points, cap.penalties, cap.bonuses )) self._log_history(cur, cap.id, "created", cap.discovered_by, f"Discovered: {cap.name}") conn.commit() return True except psycopg2.IntegrityError: return False def update_capability(self, cap_id: str, updates: Dict[str, Any], agent: str = "") -> bool: """Update a capability's fields.""" if not updates: return False # Handle JSON fields json_fields = ['parameters', 'limitations', 'examples', 'issues_found'] for fld in json_fields: if fld in updates and isinstance(updates[fld], list): updates[fld] = json.dumps(updates[fld]) updates['updated_at'] = datetime.now().isoformat() set_clause = ", ".join(f"{k} = %s" for k in updates.keys()) values = list(updates.values()) + [cap_id] with self._get_conn() as conn: with conn.cursor() as cur: cur.execute(f"UPDATE capability_registry SET {set_clause} WHERE id = %s", values) if cur.rowcount > 0: self._log_history(cur, cap_id, "updated", agent, json.dumps(updates, default=str)) conn.commit() return cur.rowcount > 0 def get_capability(self, cap_id: str) -> Optional[Capability]: """Get a capability by ID.""" with self._get_conn() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute("SELECT * FROM capability_registry WHERE id = %s", (cap_id,)) row = cur.fetchone() if row: return self._row_to_capability(row) return None def get_all(self) -> List[Capability]: """Get all capabilities.""" with self._get_conn() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute("SELECT * FROM capability_registry ORDER BY discovered_at DESC") return [self._row_to_capability(row) for row in cur.fetchall()] def get_by_status(self, status: str) -> List[Capability]: """Get capabilities by status.""" with self._get_conn() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( "SELECT * FROM capability_registry WHERE status = %s ORDER BY discovered_at DESC", (status,) ) return [self._row_to_capability(row) for row in cur.fetchall()] def get_by_category(self, category: str) -> List[Capability]: """Get capabilities by category.""" with self._get_conn() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( "SELECT * FROM capability_registry WHERE category = %s ORDER BY discovered_at DESC", (category,) ) return [self._row_to_capability(row) for row in cur.fetchall()] def search(self, query: str) -> List[Capability]: """Search capabilities by name or description.""" with self._get_conn() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(""" SELECT * FROM capability_registry WHERE name ILIKE %s OR description ILIKE %s ORDER BY discovered_at DESC """, (f"%{query}%", f"%{query}%")) return [self._row_to_capability(row) for row in cur.fetchall()] def get_stats(self) -> Dict[str, Any]: """Get registry statistics.""" with self._get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) FROM capability_registry") total = cur.fetchone()[0] cur.execute("SELECT status, COUNT(*) FROM capability_registry GROUP BY status") by_status = {row[0]: row[1] for row in cur.fetchall()} cur.execute("SELECT category, COUNT(*) FROM capability_registry GROUP BY category") by_category = {row[0]: row[1] for row in cur.fetchall()} cur.execute(""" SELECT COALESCE(SUM(discovery_points + implementation_points + documentation_points + integration_points + bonuses - penalties), 0) FROM capability_registry """) total_points = cur.fetchone()[0] or 0 return { "total_capabilities": total, "by_status": by_status, "by_category": by_category, "total_points": total_points } def get_history(self, cap_id: str, limit: int = 50) -> List[Dict]: """Get history for a capability.""" with self._get_conn() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(""" SELECT * FROM capability_history WHERE capability_id = %s ORDER BY timestamp DESC LIMIT %s """, (cap_id, limit)) return [dict(row) for row in cur.fetchall()] def _log_history(self, cur, cap_id: str, action: str, agent: str, details: str): """Log a history entry.""" cur.execute(""" INSERT INTO capability_history (capability_id, action, agent, details) VALUES (%s, %s, %s, %s) """, (cap_id, action, agent, details)) def _row_to_capability(self, row) -> Capability: """Convert a database row to a Capability object.""" return Capability( id=row['id'], name=row['name'], category=row['category'], description=row['description'] or "", discovered_by=row['discovered_by'] or "", discovered_at=row['discovered_at'] or "", discovery_source=row['discovery_source'] or "", discovery_confidence=row['discovery_confidence'] or 0.0, usage=row['usage'] or "", parameters=row['parameters'] if isinstance(row['parameters'], list) else json.loads(row['parameters'] or '[]'), limitations=row['limitations'] if isinstance(row['limitations'], list) else json.loads(row['limitations'] or '[]'), examples=row['examples'] if isinstance(row['examples'], list) else json.loads(row['examples'] or '[]'), documentation_url=row['documentation_url'] or "", status=row['status'] or "discovered", genesis_integration=row['genesis_integration'] or "", implemented_by=row['implemented_by'] or "", implemented_at=row['implemented_at'] or "", tests_passing=bool(row['tests_passing']), verified_by=row['verified_by'] or "", verified_at=row['verified_at'] or "", issues_found=row['issues_found'] if isinstance(row['issues_found'], list) else json.loads(row['issues_found'] or '[]'), quality_score=row['quality_score'] or 0.0, discovery_points=row['discovery_points'] or 0, implementation_points=row['implementation_points'] or 0, documentation_points=row['documentation_points'] or 0, integration_points=row['integration_points'] or 0, penalties=row['penalties'] or 0, bonuses=row['bonuses'] or 0 ) def exists(self, cap_id: str) -> bool: """Check if a capability exists.""" with self._get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT 1 FROM capability_registry WHERE id = %s", (cap_id,)) return cur.fetchone() is not None def delete(self, cap_id: str) -> bool: """Delete a capability and its history.""" with self._get_conn() as conn: with conn.cursor() as cur: # Delete history first (foreign key constraint) cur.execute("DELETE FROM capability_history WHERE capability_id = %s", (cap_id,)) cur.execute("DELETE FROM capability_registry WHERE id = %s", (cap_id,)) deleted = cur.rowcount > 0 conn.commit() return deleted # Quick test if __name__ == "__main__": print("Testing PostgreSQL Capability Registry (Elestio Core)...") registry = CapabilityRegistry() # Add test capability test_cap = Capability( id="cap_test_001", name="Test Capability", category="cli_features", description="A test capability", discovered_by="test_agent", discovered_at=datetime.now().isoformat(), discovery_points=10 ) if registry.add_capability(test_cap): print(" Added test capability") stats = registry.get_stats() print(f" Registry stats: {json.dumps(stats, indent=2)}") # Cleanup registry.delete("cap_test_001") print(" Test complete - PostgreSQL connection verified")