Blame for code/shared_blackboard.py

#!/usr/bin/env python3
"""
Shared Blackboard System
ISL-safe expert collaboration via evidence store
NO direct expert-to-expert messaging
"""

from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Evidence:
    """
    Immutable evidence entry from an expert
    Write-only, no modification after creation
    """
    expert: str
    finding: Any
    method: str
    confidence: float
    scope: str
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary"""
        return {
            "expert": self.expert,
            "finding": self.finding,
            "method": self.method,
            "confidence": self.confidence,
            "scope": self.scope,
            "timestamp": self.timestamp,
            "metadata": self.metadata
        }


class SharedBlackboard:
    """
    Shared evidence store for expert collaboration
    
    Rules:
    - Experts can ONLY write (no modification)
    - Entries are immutable
    - No expert-to-expert messaging
    - Orchestration by Grok Layer only
    """
    
    def __init__(self):
        self.evidence: List[Evidence] = []
        self.request_id: Optional[str] = None
        self._locked = False
    
    def start_request(self, request_id: str):
        """Start new request - clears previous evidence"""
        self.evidence = []
        self.request_id = request_id
        self._locked = False  # Unlock for new request
    
    def write_evidence(
        self,
        expert: str,
        finding: Any,
        method: str,
        confidence: float,
        scope: str = "general",
        metadata: Optional[Dict] = None
    ) -> Evidence:
        """
        Expert writes evidence to blackboard
        
        Args:
            expert: Expert identifier
            finding: The finding (structured data)
            method: How it was derived
            confidence: 0.0-1.0
            scope: What domain this applies to
            metadata: Additional context
        
        Returns:
            Evidence entry
        """
        if self._locked:
            raise RuntimeError("Blackboard is locked - no more writes allowed")
        
        # Validate confidence
        if not 0.0 <= confidence <= 1.0:
            raise ValueError(f"Confidence must be 0.0-1.0, got {confidence}")
        
        # Create immutable evidence
        evidence = Evidence(
            expert=expert,
            finding=finding,
            method=method,
            confidence=confidence,
            scope=scope,
            metadata=metadata or {}
        )
        
        self.evidence.append(evidence)
        return evidence
    
    def read_evidence(
        self,
        expert_filter: Optional[str] = None,
        scope_filter: Optional[str] = None
    ) -> List[Evidence]:
        """
        Read evidence from blackboard (read-only)
        
        Args:
            expert_filter: Only return evidence from this expert
            scope_filter: Only return evidence for this scope
        
        Returns:
            List of matching evidence
        """
        results = self.evidence
        
        if expert_filter:
            results = [e for e in results if e.expert == expert_filter]
        
        if scope_filter:
            results = [e for e in results if e.scope == scope_filter]
        
        return results
    
    def lock(self):
        """Lock blackboard - no more writes allowed"""
        self._locked = True
    
    def get_all_evidence(self) -> List[Dict]:
        """Get all evidence as dictionaries"""
        return [e.to_dict() for e in self.evidence]
    
    def get_summary(self) -> Dict[str, Any]:
        """Get summary of blackboard state"""
        return {
            "request_id": self.request_id,
            "total_evidence": len(self.evidence),
            "experts_contributed": len(set(e.expert for e in self.evidence)),
            "locked": self._locked,
            "evidence": self.get_all_evidence()
        }


class Arbitrator:
    """
    Arbitrates evidence from blackboard
    Implements consensus rules (NOT voting)
    """
    
    def __init__(self, blackboard: SharedBlackboard):
        self.blackboard = blackboard
    
    def aggregate_confidence(self) -> float:
        """
        Aggregate confidence from all evidence
        
        Rule: Confidence bounded by WEAKEST critical check
        NOT average, NOT voting
        """
        evidence = self.blackboard.evidence
        
        if not evidence:
            return 0.0
        
        # Find minimum confidence (weakest link)
        min_confidence = min(e.confidence for e in evidence)
        
        # If any expert has very low confidence, that dominates
        if min_confidence < 0.5:
            return min_confidence
        
        # Otherwise, use weighted average biased toward lower values
        confidences = [e.confidence for e in evidence]
        
        # Geometric mean (more conservative than arithmetic)
        product = 1.0
        for c in confidences:
            product *= c
        
        geometric_mean = product ** (1.0 / len(confidences))
        
        return geometric_mean
    
    def detect_conflicts(self) -> List[Dict[str, Any]]:
        """
        Detect conflicting evidence
        
        Returns:
            List of conflicts found
        """
        conflicts = []
        evidence = self.blackboard.evidence
        
        # Group by scope
        by_scope: Dict[str, List[Evidence]] = {}
        for e in evidence:
            if e.scope not in by_scope:
                by_scope[e.scope] = []
            by_scope[e.scope].append(e)
        
        # Check for contradictions within each scope
        for scope, scope_evidence in by_scope.items():
            if len(scope_evidence) < 2:
                continue
            
            # Check if findings contradict
            findings = [e.finding for e in scope_evidence]
            
            # Simple contradiction check (can be enhanced)
            if len(set(str(f) for f in findings)) > 1:
                conflicts.append({
                    "scope": scope,
                    "experts": [e.expert for e in scope_evidence],
                    "findings": findings,
                    "confidences": [e.confidence for e in scope_evidence]
                })
        
        return conflicts
    
    def synthesize_answer(self) -> Dict[str, Any]:
        """
        Synthesize final answer from all evidence
        
        Returns:
            Synthesized result with confidence
        """
        evidence = self.blackboard.evidence
        
        if not evidence:
            return {
                "answer": None,
                "confidence": 0.0,
                "method": "no_evidence"
            }
        
        # Aggregate confidence
        confidence = self.aggregate_confidence()
        
        # Detect conflicts
        conflicts = self.detect_conflicts()
        
        # If conflicts exist, reduce confidence
        if conflicts:
            confidence *= 0.7  # Penalty for conflicts
        
        # Get primary finding (highest confidence)
        primary = max(evidence, key=lambda e: e.confidence)
        
        return {
            "answer": primary.finding,
            "confidence": confidence,
            "method": "arbitrated",
            "primary_expert": primary.expert,
            "supporting_experts": [e.expert for e in evidence if e.expert != primary.expert],
            "conflicts": conflicts
        }


def test_blackboard():
    """Test shared blackboard system"""
    print("="*70)
    print("Testing Shared Blackboard System")
    print("="*70)
    
    # Create blackboard
    blackboard = SharedBlackboard()
    blackboard.start_request("test_001")
    
    # Math expert writes evidence
    print("\n1. Math Expert writes evidence:")
    math_evidence = blackboard.write_evidence(
        expert="math",
        finding=150.0,
        method="direct_multiplication",
        confidence=1.0,
        scope="calculation",
        metadata={"operation": "60 * 2.5"}
    )
    print(f"   Written: {math_evidence.finding} (confidence: {math_evidence.confidence})")
    
    # Logic expert writes evidence
    print("\n2. Logic Expert writes evidence:")
    logic_evidence = blackboard.write_evidence(
        expert="logic",
        finding="Units consistent: mph × hours → miles",
        method="unit_analysis",
        confidence=0.95,
        scope="validation"
    )
    print(f"   Written: {logic_evidence.finding} (confidence: {logic_evidence.confidence})")
    
    # Security expert writes evidence (lower confidence)
    print("\n3. Security Expert writes evidence:")
    security_evidence = blackboard.write_evidence(
        expert="security",
        finding="No security concerns",
        method="static_analysis",
        confidence=0.6,
        scope="validation"
    )
    print(f"   Written: {security_evidence.finding} (confidence: {security_evidence.confidence})")
    
    # Lock blackboard
    blackboard.lock()
    print("\n4. Blackboard locked")
    
    # Arbitrate
    print("\n5. Arbitration:")
    arbitrator = Arbitrator(blackboard)
    
    confidence = arbitrator.aggregate_confidence()
    print(f"   Aggregated confidence: {confidence:.3f}")
    
    conflicts = arbitrator.detect_conflicts()
    print(f"   Conflicts detected: {len(conflicts)}")
    
    result = arbitrator.synthesize_answer()
    print(f"   Final answer: {result['answer']}")
    print(f"   Final confidence: {result['confidence']:.3f}")
    print(f"   Primary expert: {result['primary_expert']}")
    
    # Summary
    print("\n6. Blackboard Summary:")
    summary = blackboard.get_summary()
    print(f"   Total evidence: {summary['total_evidence']}")
    print(f"   Experts contributed: {summary['experts_contributed']}")
    print(f"   Locked: {summary['locked']}")
    
    print("\n" + "="*70)
    print("✓ Blackboard system working!")
    print("="*70)


if __name__ == "__main__":
    test_blackboard()