#!/usr/bin/env python3 """ Genesis Blackboard Architecture ================================ Shared memory system for multi-agent coordination. Based on research: 13-57% better performance than RAG. Key Principles: - Central shared repository (the "blackboard") - Agents READ and WRITE autonomously - Agents CHOOSE tasks (no assignment) - Supports voting consensus and confidence scoring Usage: from blackboard import Blackboard, Agent bb = Blackboard() agent = Agent("researcher", bb) agent.post_finding("key_insight", {"data": "value"}, confidence=0.85) relevant = bb.query(pattern="insight") """ import json import os import time import threading import redis from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Any, Callable from dataclasses import dataclass, field, asdict from enum import Enum import hashlib # Import resilience utilities try: from circuit_breaker import CircuitBreaker, CircuitState, CircuitOpenError from retry_utils import retry_call RESILIENCE_AVAILABLE = True except ImportError: RESILIENCE_AVAILABLE = False CircuitBreaker = None CircuitState = None CircuitOpenError = Exception # Import secrets loader try: from secrets_loader import get_redis_config SECRETS_AVAILABLE = True except ImportError: SECRETS_AVAILABLE = False get_redis_config = None # Import atomic I/O for safe persistence try: from atomic_io import atomic_json_write, safe_json_read ATOMIC_IO_AVAILABLE = True except ImportError: ATOMIC_IO_AVAILABLE = False atomic_json_write = None safe_json_read = None class EntryType(Enum): """Types of blackboard entries.""" TASK = "task" # Work to be done FINDING = "finding" # Discovery/result DECISION = "decision" # Consensus reached QUESTION = "question" # Needs resolution SOLUTION = "solution" # Proposed answer CONFLICT = "conflict" # Disagreement to resolve class TaskStatus(Enum): """Status of tasks on blackboard.""" OPEN = "open" # Available for agents CLAIMED = "claimed" # Agent working on it COMPLETED = "completed" # Done BLOCKED = "blocked" # Cannot proceed @dataclass class BlackboardEntry: """A single entry on the blackboard.""" id: str entry_type: EntryType content: Dict[str, Any] author: str confidence: float = 0.5 timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) tags: List[str] = field(default_factory=list) references: List[str] = field(default_factory=list) # IDs of related entries votes: Dict[str, float] = field(default_factory=dict) # agent_id: vote_value status: Optional[TaskStatus] = None claimed_by: Optional[str] = None def to_dict(self) -> Dict: d = asdict(self) d['entry_type'] = self.entry_type.value if self.status: d['status'] = self.status.value return d @classmethod def from_dict(cls, data: Dict) -> 'BlackboardEntry': data['entry_type'] = EntryType(data['entry_type']) if data.get('status'): data['status'] = TaskStatus(data['status']) return cls(**data) class Blackboard: """ Central shared memory for multi-agent coordination. Features: - Thread-safe read/write operations - Pattern-based queries - Voting and consensus mechanisms - Persistence to disk - Subscription for real-time updates - Circuit breaker for Redis resilience - Graceful degradation to local storage """ def __init__(self, persist_path: Optional[str] = None, use_redis: bool = True, redis_config: Optional[Dict] = None, use_postgres: bool = True): self.entries: Dict[str, BlackboardEntry] = {} self.lock = threading.RLock() self.subscribers: Dict[str, List[Callable]] = {} self.persist_path = Path(persist_path) if persist_path else None # PostgreSQL for permanent audit/records self.use_postgres = use_postgres self.pg_conn = None if self.use_postgres: try: import psycopg2 from core.secrets_loader import get_postgres_config pg_config = get_postgres_config() if pg_config.is_configured: self.pg_conn = psycopg2.connect(pg_config.to_dsn()) self._init_pg_schema() print("[OK] Blackboard: PostgreSQL (Elestio) backend active") else: print("[!] Blackboard: PostgreSQL not configured, using local/redis only.") self.use_postgres = False except Exception as e: print(f"[!] Blackboard: PostgreSQL failed, using local/redis only: {e}") self.use_postgres = False # Redis Configuration with circuit breaker self.use_redis = use_redis self.redis_client = None self.namespace = "genesis:blackboard" self._redis_circuit = None # Initialize circuit breaker if available if RESILIENCE_AVAILABLE and CircuitBreaker: self._redis_circuit = CircuitBreaker( name="blackboard_redis", failure_threshold=3, recovery_timeout=30.0, half_open_requests=2, exceptions=(redis.RedisError, ConnectionError, TimeoutError) ) if self.use_redis: self._init_redis(redis_config) if not self.use_redis and self.persist_path and self.persist_path.exists(): self._load() def _init_pg_schema(self): """Initialize PostgreSQL schema for blackboard entries.""" if not self.pg_conn: return try: with self.pg_conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS blackboard_entries ( id TEXT PRIMARY KEY, entry_type TEXT NOT NULL, content JSONB NOT NULL, author TEXT NOT NULL, confidence FLOAT DEFAULT 0.5, timestamp TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP, tags TEXT[], "references" TEXT[], votes JSONB DEFAULT '{}', status TEXT, claimed_by TEXT ); CREATE INDEX IF NOT EXISTS idx_bb_type ON blackboard_entries(entry_type); CREATE INDEX IF NOT EXISTS idx_bb_author ON blackboard_entries(author); CREATE INDEX IF NOT EXISTS idx_bb_status ON blackboard_entries(status); """) self.pg_conn.commit() except Exception as e: print(f"[!] Blackboard: Failed to initialize PG schema: {e}") self.pg_conn.rollback() raise def _init_redis(self, redis_config: Optional[Dict] = None) -> None: """Initialize Redis connection with fallback chain.""" try: # Priority 1: Use provided config if redis_config: pass # Use as-is # Priority 2: Use secrets_loader elif SECRETS_AVAILABLE and get_redis_config: secrets_config = get_redis_config() if secrets_config.is_configured: redis_config = secrets_config.to_dict() # Priority 3: Fall back to config file if not redis_config: config_path = r"E:\genesis-system\genesis_config.json" if os.path.exists(config_path): with open(config_path, 'r') as f: full_config = json.load(f) redis_config = full_config.get("redis") if redis_config: self.redis_client = redis.Redis( host=redis_config["host"], port=redis_config["port"], password=redis_config.get("password"), ssl=redis_config.get("ssl", False), decode_responses=True, socket_timeout=5 ) self.redis_client.ping() print("[OK] Blackboard: Redis backend connected.") else: print("[!] Blackboard: No Redis config found, using local storage.") self.use_redis = False except Exception as e: print(f"[!] Blackboard Redis Error: {e}") print("[!] Blackboard: Falling back to local storage (graceful degradation).") self.use_redis = False if self._redis_circuit: self._redis_circuit.record_failure(e) def _redis_available(self) -> bool: """Check if Redis is available (considering circuit breaker).""" if not self.use_redis or not self.redis_client: return False if self._redis_circuit and not self._redis_circuit.is_available: return False return True def _redis_op(self, operation: Callable, fallback: Callable = None) -> Any: """Execute Redis operation with circuit breaker protection.""" if not self._redis_available(): if fallback: return fallback() return None try: if self._redis_circuit: result = self._redis_circuit.call(operation) else: result = operation() return result except (redis.RedisError, CircuitOpenError) as e: print(f"[!] Blackboard Redis op failed: {e}") if fallback: return fallback() return None def _generate_id(self, content: Dict) -> str: """Generate unique ID for entry.""" timestamp = datetime.now().isoformat() content_str = json.dumps(content, sort_keys=True) hash_input = f"{timestamp}:{content_str}" return hashlib.sha256(hash_input.encode()).hexdigest()[:12] def write(self, entry_type: EntryType, content: Dict[str, Any], author: str, confidence: float = 0.5, tags: List[str] = None, references: List[str] = None, status: TaskStatus = None) -> str: """ Write an entry to the blackboard. Returns: Entry ID """ with self.lock: entry_id = self._generate_id(content) entry = BlackboardEntry( id=entry_id, entry_type=entry_type, content=content, author=author, confidence=confidence, tags=tags or [], references=references or [], status=status ) if self.use_redis: self.redis_client.set( f"{self.namespace}:{entry_id}", json.dumps(entry.to_dict()) ) else: self.entries[entry_id] = entry self._persist() self._notify_subscribers(entry_type, entry) return entry_id def read(self, entry_id: str) -> Optional[BlackboardEntry]: """Read a specific entry by ID.""" if self.use_redis: data = self.redis_client.get(f"{self.namespace}:{entry_id}") if data: return BlackboardEntry.from_dict(json.loads(data)) return None with self.lock: return self.entries.get(entry_id) def query(self, entry_type: EntryType = None, pattern: str = None, tags: List[str] = None, author: str = None, min_confidence: float = 0.0, status: TaskStatus = None, limit: int = 100) -> List[BlackboardEntry]: """ Query entries matching criteria. Args: entry_type: Filter by type pattern: Search in content (simple substring match) tags: Must have all these tags author: Filter by author min_confidence: Minimum confidence threshold status: Filter by task status limit: Maximum results Returns: List of matching entries, newest first """ results = [] if self.use_redis: # Note: For scale, we'd use Redis Search. # For now, we scan keys to maintain compatibility with the existing patterns. for key in self.redis_client.scan_iter(f"{self.namespace}:*"): data = self.redis_client.get(key) if not data: continue entry = BlackboardEntry.from_dict(json.loads(data)) # Apply filters if entry_type and entry.entry_type != entry_type: continue if pattern: if pattern.lower() not in json.dumps(entry.content).lower(): continue if tags and not all(t in entry.tags for t in tags): continue if author and entry.author != author: continue if entry.confidence < min_confidence: continue if status and entry.status != status: continue results.append(entry) else: with self.lock: for entry in self.entries.values(): if entry_type and entry.entry_type != entry_type: continue if pattern: if pattern.lower() not in json.dumps(entry.content).lower(): continue if tags and not all(t in entry.tags for t in tags): continue if author and entry.author != author: continue if entry.confidence < min_confidence: continue if status and entry.status != status: continue results.append(entry) # Sort by timestamp descending results.sort(key=lambda e: e.timestamp, reverse=True) return results[:limit] def get_open_tasks(self) -> List[BlackboardEntry]: """Get all open tasks available for claiming.""" return self.query(entry_type=EntryType.TASK, status=TaskStatus.OPEN) def claim_task(self, task_id: str, agent_id: str) -> bool: """ Claim an open task for an agent. Returns True if successfully claimed. """ if self.use_redis: # Atomic claim using Redis transaction with self.redis_client.pipeline() as pipe: try: key = f"{self.namespace}:{task_id}" pipe.watch(key) data = pipe.get(key) if not data: return False entry_dict = json.loads(data) if entry_dict.get('status') != TaskStatus.OPEN.value: return False entry_dict['status'] = TaskStatus.CLAIMED.value entry_dict['claimed_by'] = agent_id pipe.multi() pipe.set(key, json.dumps(entry_dict)) pipe.execute() return True except redis.WatchError: return False with self.lock: entry = self.entries.get(task_id) if not entry or entry.status != TaskStatus.OPEN: return False entry.status = TaskStatus.CLAIMED entry.claimed_by = agent_id self._persist() return True def log_proof_of_work(self, task_id: str, agent_id: str, actions: List[Dict], evidence: str = ""): """ Log verifiable proof of work for a task. Critical for 'Black Belt' mastery audit. """ pow_entry = { "task_id": task_id, "agent_id": agent_id, "timestamp": datetime.now().isoformat(), "actions": actions, "evidence": evidence, "status": "verifiable" } self.write( entry_type=EntryType.FINDING, content=pow_entry, author=agent_id, tags=["proof_of_work", "audit_trail"], references=[task_id] ) print(f"[POW] PoW logged for {task_id} by {agent_id}") def complete_task(self, task_id: str, agent_id: str, result: Dict = None) -> bool: """Mark a task as completed.""" if self.use_redis: key = f"{self.namespace}:{task_id}" data = self.redis_client.get(key) if not data: return False entry_dict = json.loads(data) if entry_dict.get('claimed_by') != agent_id: return False entry_dict['status'] = TaskStatus.COMPLETED.value if result: if 'content' not in entry_dict: entry_dict['content'] = {} entry_dict['content']['result'] = result self.redis_client.set(key, json.dumps(entry_dict)) return True with self.lock: entry = self.entries.get(task_id) if not entry or entry.claimed_by != agent_id: return False entry.status = TaskStatus.COMPLETED if result: entry.content["result"] = result self._persist() return True def vote(self, entry_id: str, agent_id: str, vote_value: float) -> bool: """ Cast a vote on an entry. Args: entry_id: Entry to vote on agent_id: Voting agent vote_value: -1.0 to 1.0 (disagree to agree) Returns: True if vote recorded """ if self.use_redis: key = f"{self.namespace}:{entry_id}" data = self.redis_client.get(key) if not data: return False entry_dict = json.loads(data) if 'votes' not in entry_dict: entry_dict['votes'] = {} entry_dict['votes'][agent_id] = max(-1.0, min(1.0, vote_value)) self.redis_client.set(key, json.dumps(entry_dict)) return True with self.lock: entry = self.entries.get(entry_id) if not entry: return False entry.votes[agent_id] = max(-1.0, min(1.0, vote_value)) self._persist() return True def get_consensus(self, entry_id: str) -> Dict[str, Any]: """ Calculate consensus on an entry. Returns: { "vote_count": int, "average": float, "consensus_reached": bool, # True if avg > 0.6 and count >= 2 "votes": dict } """ with self.lock: entry = self.entries.get(entry_id) if not entry or not entry.votes: return {"vote_count": 0, "average": 0.0, "consensus_reached": False, "votes": {}} votes = list(entry.votes.values()) avg = sum(votes) / len(votes) return { "vote_count": len(votes), "average": avg, "consensus_reached": avg > 0.6 and len(votes) >= 2, "votes": entry.votes } def subscribe(self, entry_type: EntryType, callback: Callable[[BlackboardEntry], None]): """Subscribe to new entries of a type.""" type_key = entry_type.value if type_key not in self.subscribers: self.subscribers[type_key] = [] self.subscribers[type_key].append(callback) def _notify_subscribers(self, entry_type: EntryType, entry: BlackboardEntry): """Notify subscribers of new entry.""" type_key = entry_type.value for callback in self.subscribers.get(type_key, []): try: callback(entry) except Exception as e: print(f"Subscriber error: {e}") def _persist(self): """Save blackboard to disk.""" if not self.persist_path: return data = { "version": "1.0", "updated": datetime.now().isoformat(), "entries": {k: v.to_dict() for k, v in self.entries.items()} } self.persist_path.parent.mkdir(parents=True, exist_ok=True) if ATOMIC_IO_AVAILABLE and atomic_json_write: atomic_json_write(self.persist_path, data) else: with open(self.persist_path, 'w') as f: json.dump(data, f, indent=2) def _load(self): """Load blackboard from disk.""" try: with open(self.persist_path, 'r') as f: data = json.load(f) self.entries = { k: BlackboardEntry.from_dict(v) for k, v in data.get("entries", {}).items() } except Exception as e: print(f"Error loading blackboard: {e}") def stats(self) -> Dict[str, Any]: """Get blackboard statistics.""" if self.use_redis: by_type = {} by_status = {} total_confidence = 0.0 count = 0 for key in self.redis_client.scan_iter(f"{self.namespace}:*"): data = self.redis_client.get(key) if not data: continue entry = json.loads(data) t = entry.get('entry_type') by_type[t] = by_type.get(t, 0) + 1 s = entry.get('status') if s: by_status[s] = by_status.get(s, 0) + 1 total_confidence += entry.get('confidence', 0.5) count += 1 return { "backend": "redis", "total_entries": count, "by_type": by_type, "by_status": by_status, "average_confidence": total_confidence / count if count else 0.0 } with self.lock: by_type = {} by_status = {} total_confidence = 0.0 for entry in self.entries.values(): t = entry.entry_type.value by_type[t] = by_type.get(t, 0) + 1 if entry.status: s = entry.status.value by_status[s] = by_status.get(s, 0) + 1 total_confidence += entry.confidence return { "backend": "local", "total_entries": len(self.entries), "by_type": by_type, "by_status": by_status, "average_confidence": total_confidence / len(self.entries) if self.entries else 0.0 } class VotingStrategy(Enum): """Voting strategies for different decision types.""" MAJORITY = "majority" # > 50% agreement SUPERMAJORITY = "supermajority" # >= 67% agreement UNANIMOUS = "unanimous" # 100% agreement WEIGHTED = "weighted" # Confidence-weighted QUORUM = "quorum" # Minimum participation required @dataclass class VotingConfig: """Configuration for a voting session.""" strategy: VotingStrategy = VotingStrategy.MAJORITY quorum: int = 2 # Minimum voters needed timeout_seconds: int = 300 # 5 minute default allow_abstain: bool = True weight_by_confidence: bool = False require_domain_expertise: bool = False @dataclass class VoteRecord: """Individual vote record.""" agent_id: str value: float # -1.0 to 1.0 confidence: float = 0.5 # Voter's confidence in their vote domain_expertise: float = 0.5 # Agent's expertise in this domain rationale: str = "" timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) class VotingConsensus: """ Advanced voting consensus system for multi-agent decisions. Research-backed improvements (13.2% reasoning improvement): - Majority voting for lower-stakes decisions - Unanimous for irreversible actions - Confidence scoring for conflict resolution - Weighted voting based on domain expertise """ def __init__(self, blackboard: 'Blackboard'): self.blackboard = blackboard self.session_namespace = "genesis:voting" # Local cache for non-redis mode self.active_sessions: Dict[str, Dict] = {} self.completed_sessions: List[Dict] = [] def start_vote(self, topic: str, options: List[str], config: VotingConfig = None, context: Dict = None) -> str: """ Start a new voting session. Args: topic: What we're voting on options: Available choices (or empty for agree/disagree) config: Voting configuration context: Additional context for voters Returns: Session ID """ config = config or VotingConfig() session_id = hashlib.sha256( f"{topic}{datetime.now().isoformat()}".encode() ).hexdigest()[:12] config_dict = asdict(config) config_dict['strategy'] = config.strategy.value # Convert enum to string session_data = { "id": session_id, "topic": topic, "options": options or ["agree", "disagree"], "config": config_dict, "context": context or {}, "votes": {}, "started_at": datetime.now().isoformat(), "status": "active" } if self.blackboard.use_redis: self.blackboard.redis_client.set( f"{self.session_namespace}:{session_id}", json.dumps(session_data) ) else: session_data["config_obj"] = config self.active_sessions[session_id] = session_data # Also post to blackboard for visibility self.blackboard.write( entry_type=EntryType.DECISION, content={ "session_id": session_id, "topic": topic, "options": options, "status": "voting" }, author="voting_system", confidence=0.5 ) return session_id def cast_vote(self, session_id: str, agent_id: str, choice: str, confidence: float = 0.5, domain_expertise: float = 0.5, rationale: str = "") -> bool: """Cast a vote in an active session.""" session = self.get_session(session_id) if not session or session["status"] != "active": return False options = session["options"] vote = VoteRecord( agent_id=agent_id, value=1.0 if choice == options[0] else (-1.0 if len(options) > 1 and choice == options[1] else 0.0), confidence=confidence, domain_expertise=domain_expertise, rationale=rationale ) if self.blackboard.use_redis: key = f"{self.session_namespace}:{session_id}" for _ in range(3): # Retry up to 3 times on collision with self.blackboard.redis_client.pipeline() as pipe: try: pipe.watch(key) data = pipe.get(key) if not data: return False session = json.loads(data) if "votes" not in session: session["votes"] = {} session["votes"][agent_id] = { "choice": choice, "record": asdict(vote) } pipe.multi() pipe.set(key, json.dumps(session)) pipe.execute() return True except redis.WatchError: time.sleep(0.1) continue return False # Local non-redis logic session["votes"][agent_id] = { "choice": choice, "record": asdict(vote) } return True def calculate_result(self, session_id: str) -> Dict[str, Any]: """ Calculate voting result based on strategy. Returns: { "decision": str, # Winning option or "no_consensus" "passed": bool, # Whether vote passed "vote_counts": dict, # Count per option "total_votes": int, "quorum_met": bool, "strategy_result": dict, # Strategy-specific details "confidence_score": float } """ session = self.get_session(session_id) if not session: return {"error": "Session not found"} votes = session["votes"] # Restore config obj config_dict = session["config"] config = VotingConfig( strategy=VotingStrategy(config_dict['strategy']), quorum=config_dict.get('quorum', 2), timeout_seconds=config_dict.get('timeout_seconds', 300), allow_abstain=config_dict.get('allow_abstain', True), weight_by_confidence=config_dict.get('weight_by_confidence', False), require_domain_expertise=config_dict.get('require_domain_expertise', False) ) options = session["options"] # Count votes per option vote_counts = {opt: 0 for opt in options} vote_counts["abstain"] = 0 weighted_scores = {opt: 0.0 for opt in options} total_confidence = 0.0 for agent_id, vote_data in votes.items(): choice = vote_data["choice"] record = vote_data["record"] if choice in vote_counts: vote_counts[choice] += 1 # Calculate weighted score weight = 1.0 if config.weight_by_confidence: weight *= record["confidence"] if config.require_domain_expertise: weight *= record["domain_expertise"] if choice in weighted_scores: weighted_scores[choice] += weight total_confidence += record["confidence"] total_votes = len(votes) non_abstain_votes = total_votes - vote_counts.get("abstain", 0) quorum_met = total_votes >= config.quorum # Apply voting strategy result = self._apply_strategy( config.strategy, options, vote_counts, weighted_scores, non_abstain_votes ) passed = quorum_met and result["passed"] decision = result["winner"] if passed else "no_consensus" confidence_score = total_confidence / total_votes if total_votes > 0 else 0.0 return { "session_id": session_id, "topic": session["topic"], "decision": decision, "passed": passed, "vote_counts": vote_counts, "weighted_scores": weighted_scores, "total_votes": total_votes, "quorum_met": quorum_met, "quorum_required": config.quorum, "strategy": config.strategy.value, "strategy_result": result, "confidence_score": round(confidence_score, 3) } def _apply_strategy(self, strategy: VotingStrategy, options: List[str], counts: Dict[str, int], weighted: Dict[str, float], total: int) -> Dict: """Apply voting strategy to determine result.""" if total == 0: return {"passed": False, "winner": None, "reason": "No votes cast"} if strategy == VotingStrategy.MAJORITY: # Simple majority: > 50% for opt in options: if counts.get(opt, 0) / total > 0.5: return {"passed": True, "winner": opt, "percentage": counts[opt] / total} return {"passed": False, "winner": None, "reason": "No majority reached"} elif strategy == VotingStrategy.SUPERMAJORITY: # Supermajority: >= 67% for opt in options: if counts.get(opt, 0) / total >= 0.67: return {"passed": True, "winner": opt, "percentage": counts[opt] / total} return {"passed": False, "winner": None, "reason": "No supermajority reached"} elif strategy == VotingStrategy.UNANIMOUS: # All votes must be the same for opt in options: if counts.get(opt, 0) == total: return {"passed": True, "winner": opt, "percentage": 1.0} return {"passed": False, "winner": None, "reason": "Not unanimous"} elif strategy == VotingStrategy.WEIGHTED: # Winner is highest weighted score winner = max(weighted.keys(), key=lambda x: weighted.get(x, 0)) total_weight = sum(weighted.values()) if total_weight > 0 and weighted[winner] / total_weight > 0.5: return {"passed": True, "winner": winner, "weighted_score": weighted[winner]} return {"passed": False, "winner": None, "reason": "No weighted majority"} elif strategy == VotingStrategy.QUORUM: # Just needs quorum, then plurality wins winner = max(options, key=lambda x: counts.get(x, 0)) return {"passed": True, "winner": winner, "count": counts[winner]} return {"passed": False, "winner": None, "reason": "Unknown strategy"} def close_vote(self, session_id: str) -> Dict: """Close voting and finalize result.""" result = self.calculate_result(session_id) session = self.get_session(session_id) if session and session["status"] == "active": session["status"] = "closed" session["result"] = result session["closed_at"] = datetime.now().isoformat() if self.blackboard.use_redis: # Remove active, store in completed (optional - for now just update status) self.blackboard.redis_client.set( f"{self.session_namespace}:{session_id}", json.dumps(session) ) else: self.completed_sessions.append(session) if session_id in self.active_sessions: del self.active_sessions[session_id] return result def get_session(self, session_id: str) -> Optional[Dict]: """Get a voting session's details.""" if self.blackboard.use_redis: data = self.blackboard.redis_client.get(f"{self.session_namespace}:{session_id}") return json.loads(data) if data else None return self.active_sessions.get(session_id) def list_active(self) -> List[Dict]: """List all active voting sessions.""" active = [] if self.blackboard.use_redis: for key in self.blackboard.redis_client.scan_iter(f"{self.session_namespace}:*"): data = self.blackboard.redis_client.get(key) if data: s = json.loads(data) if s.get("status") == "active": active.append({"id": s["id"], "topic": s["topic"], "votes": len(s["votes"])}) else: for s in self.active_sessions.values(): active.append({"id": s["id"], "topic": s["topic"], "votes": len(s["votes"])}) return active class Agent: """ An agent that interacts with the blackboard. Agents can: - Post findings and solutions - Claim and complete tasks - Vote on entries - Query for relevant information """ def __init__(self, agent_id: str, blackboard: Blackboard, domain: str = "general"): self.agent_id = agent_id self.blackboard = blackboard self.domain = domain def post_finding(self, key: str, data: Any, confidence: float = 0.5, tags: List[str] = None) -> str: """Post a finding to the blackboard.""" return self.blackboard.write( entry_type=EntryType.FINDING, content={"key": key, "data": data}, author=self.agent_id, confidence=confidence, tags=tags or [self.domain] ) def post_solution(self, problem_id: str, solution: Any, confidence: float = 0.5) -> str: """Post a solution to a problem/question.""" return self.blackboard.write( entry_type=EntryType.SOLUTION, content={"solution": solution}, author=self.agent_id, confidence=confidence, references=[problem_id] ) def post_task(self, description: str, requirements: Dict = None, tags: List[str] = None) -> str: """Post a new task to the blackboard.""" return self.blackboard.write( entry_type=EntryType.TASK, content={"description": description, "requirements": requirements or {}}, author=self.agent_id, tags=tags or [self.domain], status=TaskStatus.OPEN ) def claim_available_task(self, tags: List[str] = None) -> Optional[str]: """ Find and claim an open task matching criteria. Returns: Task ID if claimed, None if no tasks available """ tasks = self.blackboard.query( entry_type=EntryType.TASK, status=TaskStatus.OPEN, tags=tags ) for task in tasks: if self.blackboard.claim_task(task.id, self.agent_id): return task.id return None def complete_task(self, task_id: str, result: Dict = None) -> bool: """Complete a task I claimed.""" return self.blackboard.complete_task(task_id, self.agent_id, result) def vote_on(self, entry_id: str, agree: bool) -> bool: """Vote on an entry (agree=True: 1.0, agree=False: -1.0).""" return self.blackboard.vote(entry_id, self.agent_id, 1.0 if agree else -1.0) def find_relevant(self, pattern: str, min_confidence: float = 0.3) -> List[BlackboardEntry]: """Find entries relevant to a pattern.""" return self.blackboard.query(pattern=pattern, min_confidence=min_confidence) # CLI Interface if __name__ == "__main__": import sys # Use persistent storage in genesis system bb_path = Path("/mnt/e/genesis-system/blackboard_state.json") bb = Blackboard(persist_path=str(bb_path)) if len(sys.argv) < 2: print(""" Genesis Blackboard ================== Commands: stats Show blackboard statistics tasks List open tasks findings List recent findings post Post an entry (types: task, finding, question) vote Vote on entry (-1 to 1) consensus Check consensus on entry Examples: python blackboard.py stats python blackboard.py post task '{"description": "Implement voting"}' python blackboard.py vote abc123 0.8 """) sys.exit(0) command = sys.argv[1] if command == "stats": stats = bb.stats() print(f"Total Entries: {stats['total_entries']}") print(f"By Type: {stats['by_type']}") print(f"By Status: {stats['by_status']}") print(f"Avg Confidence: {stats['average_confidence']:.2f}") elif command == "tasks": tasks = bb.get_open_tasks() if not tasks: print("No open tasks") else: for task in tasks: print(f"[{task.id}] {task.content.get('description', 'No description')}") elif command == "findings": findings = bb.query(entry_type=EntryType.FINDING, limit=10) for f in findings: print(f"[{f.id}] ({f.confidence:.2f}) {f.content.get('key', 'Unknown')}: {f.content.get('data', '')[:50]}") elif command == "post" and len(sys.argv) >= 4: entry_type = EntryType(sys.argv[2]) content = json.loads(sys.argv[3]) # Tasks should be open by default status = TaskStatus.OPEN if entry_type == EntryType.TASK else None entry_id = bb.write( entry_type=entry_type, content=content, author="cli", confidence=0.5, status=status ) print(f"Created entry: {entry_id}") elif command == "vote" and len(sys.argv) >= 4: entry_id = sys.argv[2] value = float(sys.argv[3]) if bb.vote(entry_id, "cli", value): print(f"Vote recorded: {value}") else: print("Entry not found") elif command == "consensus" and len(sys.argv) >= 3: entry_id = sys.argv[2] result = bb.get_consensus(entry_id) print(f"Votes: {result['vote_count']}") print(f"Average: {result['average']:.2f}") print(f"Consensus Reached: {result['consensus_reached']}") else: print(f"Unknown command: {command}") sys.exit(1)