tribal_cognition.py
1 """ 2 Tribal Cognition - Multi-User Collective Sense-Making 3 4 Implements the tribal dynamics pattern: 5 - Individual cognitive fingerprints per user 6 - Collective graph (shared gravity wells, precedents) 7 - Campfire synchronization between individual and collective 8 - Escalation system (Moses pattern) for edge cases 9 - Role assignment (elder, working, young) 10 11 ## Organic Parallel: Tribal Dynamics 12 13 In tribal societies, knowledge flowed around the campfire: 14 - Elders (long memory, pattern recognition, judgment) 15 - Working-age (current execution, bridge old and new) 16 - Young (fresh eyes, questions assumptions) 17 18 The Moses pattern: Most decisions are local. Edge cases escalate. 19 Precedents flow down from elders to become tribal law. 20 """ 21 22 from dataclasses import dataclass, field 23 from datetime import datetime, timedelta 24 from typing import Optional, List, Dict, Set, Tuple, Any, Callable 25 from enum import Enum 26 from collections import defaultdict 27 import math 28 import json 29 import hashlib 30 31 32 class TribalRole(Enum): 33 """Role in the tribal hierarchy.""" 34 FOUNDER = "founder" # Ultimate elder, origin memory 35 ELDER = "elder" # Long tenure, earned judgment 36 WORKING = "working" # Active execution, bridge role 37 YOUNG = "young" # Fresh perspective, learning 38 39 40 @dataclass 41 class GravityWell: 42 """A topic that attracts attention.""" 43 concept: str 44 mass: float = 0.5 # Attraction strength (0-1) 45 pull_radius: float = 0.3 # Influence on related concepts 46 activation_count: int = 0 47 last_activated: datetime = field(default_factory=datetime.now) 48 contributors: List[str] = field(default_factory=list) # Who strengthened this 49 50 def activate(self, user_id: str = None): 51 """Strengthen this gravity well.""" 52 self.last_activated = datetime.now() 53 self.activation_count += 1 54 self.mass = min(0.95, self.mass + 0.02) 55 if user_id and user_id not in self.contributors: 56 self.contributors.append(user_id) 57 58 def decay(self, days: float): 59 """Apply time-based decay.""" 60 decay_factor = 0.99 ** days 61 self.mass = max(0.1, self.mass * decay_factor) 62 63 64 @dataclass 65 class Precedent: 66 """ 67 A judgment from an elder that becomes tribal law. 68 69 Once an elder makes a judgment, it can be applied locally 70 without re-escalating. 71 """ 72 id: str 73 judge_id: str # Who made the judgment 74 judge_role: TribalRole 75 judgment_date: datetime 76 77 # The case 78 original_question: str 79 context: str 80 pattern_signature: str # For matching similar cases 81 82 # The judgment 83 decision: str 84 reasoning: str 85 86 # Scope 87 applies_to_patterns: List[str] # Pattern signatures this covers 88 supersedes: Optional[str] = None # Previous precedent this updates 89 90 # Usage tracking 91 times_applied: int = 0 92 times_challenged: int = 0 93 last_applied: Optional[datetime] = None 94 95 def apply(self): 96 """Record that this precedent was applied.""" 97 self.times_applied += 1 98 self.last_applied = datetime.now() 99 100 101 @dataclass 102 class UserCognitiveState: 103 """An individual user's sense-making model.""" 104 user_id: str 105 display_name: str 106 join_date: datetime 107 role: TribalRole = TribalRole.YOUNG 108 109 # Personal gravity wells 110 gravity_wells: Dict[str, GravityWell] = field(default_factory=dict) 111 112 # Personal pattern preferences (what patterns resonate) 113 pattern_preferences: Dict[str, float] = field(default_factory=dict) 114 115 # Altitude distribution (how they tend to think) 116 altitude_distribution: Dict[str, float] = field(default_factory=lambda: { 117 'philosophical': 0.25, 118 'strategic': 0.25, 119 'tactical': 0.25, 120 'operational': 0.25 121 }) 122 123 # Trust in collective vs self (0 = all self, 1 = all collective) 124 collective_weight: float = 0.5 125 126 # Interaction tracking 127 last_active: datetime = field(default_factory=datetime.now) 128 last_campfire: Optional[datetime] = None 129 escalations_made: int = 0 130 precedents_set: int = 0 131 precedents_received: int = 0 132 133 @property 134 def tenure_days(self) -> int: 135 """Days since joining.""" 136 return (datetime.now() - self.join_date).days 137 138 def get_gravity_for_concept(self, concept: str) -> float: 139 """Get gravity strength for a concept.""" 140 if concept in self.gravity_wells: 141 return self.gravity_wells[concept].mass 142 # Check related concepts 143 total = 0.0 144 for well in self.gravity_wells.values(): 145 if concept.lower() in well.concept.lower() or well.concept.lower() in concept.lower(): 146 total += well.mass * well.pull_radius 147 return min(1.0, total) 148 149 def add_gravity_well(self, concept: str, initial_mass: float = 0.5): 150 """Add or strengthen a gravity well.""" 151 if concept in self.gravity_wells: 152 self.gravity_wells[concept].activate(self.user_id) 153 else: 154 self.gravity_wells[concept] = GravityWell( 155 concept=concept, 156 mass=initial_mass, 157 contributors=[self.user_id] 158 ) 159 160 161 @dataclass 162 class CollectiveGraph: 163 """ 164 The organization's shared understanding. 165 166 This is the "tribal memory" that transcends individuals. 167 """ 168 # Shared gravity wells (what the tribe cares about) 169 collective_wells: Dict[str, GravityWell] = field(default_factory=dict) 170 171 # Precedents (judgments that became law) 172 precedents: List[Precedent] = field(default_factory=list) 173 174 # Founding principles (invariants from founder) 175 founding_principles: List[str] = field(default_factory=list) 176 177 # Campfire history 178 campfire_history: List[Dict] = field(default_factory=list) 179 180 # Last update 181 updated_at: datetime = field(default_factory=datetime.now) 182 183 def get_well_strength(self, concept: str) -> float: 184 """Get collective importance of a concept.""" 185 if concept in self.collective_wells: 186 return self.collective_wells[concept].mass 187 return 0.0 188 189 def find_precedent(self, pattern_signature: str) -> Optional[Precedent]: 190 """Find relevant precedent for a pattern.""" 191 for precedent in reversed(self.precedents): # Most recent first 192 if pattern_signature in precedent.applies_to_patterns: 193 return precedent 194 # Fuzzy matching on signature 195 if self._pattern_similarity(pattern_signature, precedent.pattern_signature) > 0.7: 196 return precedent 197 return None 198 199 def _pattern_similarity(self, p1: str, p2: str) -> float: 200 """Compute similarity between pattern signatures.""" 201 words1 = set(p1.lower().split()) 202 words2 = set(p2.lower().split()) 203 if not words1 or not words2: 204 return 0.0 205 intersection = words1 & words2 206 union = words1 | words2 207 return len(intersection) / len(union) 208 209 def add_precedent(self, precedent: Precedent): 210 """Add a new precedent.""" 211 self.precedents.append(precedent) 212 self.updated_at = datetime.now() 213 214 215 @dataclass 216 class EscalationRequest: 217 """A request to escalate a decision to a higher level.""" 218 id: str 219 from_user: str 220 to_judge: str 221 timestamp: datetime 222 223 # The question 224 question: str 225 context: str 226 pattern_signature: str 227 228 # Why escalating 229 reason: str 230 confidence_score: float # How confident was the user (low = escalate) 231 232 # Resolution 233 resolved: bool = False 234 resolution: Optional[str] = None 235 became_precedent: bool = False 236 237 238 @dataclass 239 class CampfireReport: 240 """Results of a campfire synchronization.""" 241 timestamp: datetime 242 participants: List[str] 243 244 # Collective updates 245 new_collective_wells: List[str] 246 strengthened_wells: List[str] 247 decayed_wells: List[str] 248 249 # Individual updates 250 individual_updates: Dict[str, Dict] # user_id -> updates 251 252 # New precedents 253 new_precedents: List[str] 254 255 # Escalations generated 256 escalations: List[str] 257 258 259 class RoleAssigner: 260 """Assigns roles based on tenure and demonstrated judgment.""" 261 262 # Tenure thresholds in days 263 YOUNG_MAX_DAYS = 90 264 WORKING_MAX_DAYS = 730 # 2 years 265 266 # Judgment thresholds for promotion 267 ELDER_PRECEDENTS_THRESHOLD = 5 268 ELDER_ALIGNMENT_THRESHOLD = 0.8 269 WORKING_COMPETENCE_THRESHOLD = 0.7 270 271 def assign_role( 272 self, 273 user: UserCognitiveState, 274 collective: CollectiveGraph, 275 is_founder: bool = False 276 ) -> TribalRole: 277 """Determine user's role in the tribal hierarchy.""" 278 279 if is_founder: 280 return TribalRole.FOUNDER 281 282 # Base assignment by tenure 283 if user.tenure_days < self.YOUNG_MAX_DAYS: 284 base_role = TribalRole.YOUNG 285 elif user.tenure_days < self.WORKING_MAX_DAYS: 286 base_role = TribalRole.WORKING 287 else: 288 base_role = TribalRole.ELDER 289 290 # Check for promotion based on demonstrated judgment 291 if base_role == TribalRole.WORKING: 292 if self._has_elder_qualities(user, collective): 293 return TribalRole.ELDER 294 295 if base_role == TribalRole.YOUNG: 296 if self._has_working_qualities(user, collective): 297 return TribalRole.WORKING 298 299 return base_role 300 301 def _has_elder_qualities(self, user: UserCognitiveState, collective: CollectiveGraph) -> bool: 302 """Has this user demonstrated elder-level judgment?""" 303 # Precedents they set 304 if user.precedents_set >= self.ELDER_PRECEDENTS_THRESHOLD: 305 return True 306 307 # Alignment with collective 308 alignment = self._compute_collective_alignment(user, collective) 309 if alignment > self.ELDER_ALIGNMENT_THRESHOLD: 310 return True 311 312 return False 313 314 def _has_working_qualities(self, user: UserCognitiveState, collective: CollectiveGraph) -> bool: 315 """Has this user demonstrated working-level competence?""" 316 # Basic competence: has gravity wells that align with collective 317 alignment = self._compute_collective_alignment(user, collective) 318 return alignment > self.WORKING_COMPETENCE_THRESHOLD 319 320 def _compute_collective_alignment(self, user: UserCognitiveState, collective: CollectiveGraph) -> float: 321 """How well do user's gravity wells align with collective?""" 322 if not user.gravity_wells or not collective.collective_wells: 323 return 0.5 # Neutral 324 325 user_concepts = set(user.gravity_wells.keys()) 326 collective_concepts = set(collective.collective_wells.keys()) 327 328 if not user_concepts: 329 return 0.5 330 331 overlap = user_concepts & collective_concepts 332 return len(overlap) / len(user_concepts) 333 334 335 class EscalationManager: 336 """ 337 Implements the Moses pattern: local decisions with escalation for edge cases. 338 """ 339 340 def __init__(self, collective: CollectiveGraph): 341 self.collective = collective 342 self.pending_escalations: List[EscalationRequest] = [] 343 self.resolved_escalations: List[EscalationRequest] = [] 344 345 def should_escalate( 346 self, 347 question: str, 348 context: str, 349 user: UserCognitiveState, 350 confidence: float, 351 stakes: str = "medium" 352 ) -> Tuple[bool, str]: 353 """ 354 Should this decision be escalated to someone higher? 355 356 Returns: (should_escalate, reason) 357 """ 358 pattern = self._compute_pattern_signature(question, context) 359 360 # Check for existing precedent 361 precedent = self.collective.find_precedent(pattern) 362 if precedent: 363 return False, f"Precedent exists: {precedent.id}" 364 365 # Check decision confidence 366 if confidence > 0.8: 367 return False, "High confidence - decide locally" 368 369 # New user + high stakes = escalate 370 if user.tenure_days < 90 and stakes == "high": 371 return True, "New user, high stakes - seek elder judgment" 372 373 # Low confidence = escalate 374 if confidence < 0.4: 375 return True, "Low confidence - seek guidance" 376 377 # Check if multiple gravity wells conflict 378 conflicting_wells = self._find_conflicting_wells(question, user) 379 if len(conflicting_wells) > 2: 380 return True, f"Conflicting gravity wells: {conflicting_wells}" 381 382 return False, "Can decide locally" 383 384 def create_escalation( 385 self, 386 question: str, 387 context: str, 388 from_user: UserCognitiveState, 389 to_user: UserCognitiveState, 390 reason: str, 391 confidence: float 392 ) -> EscalationRequest: 393 """Create an escalation request.""" 394 escalation = EscalationRequest( 395 id=self._generate_id(question), 396 from_user=from_user.user_id, 397 to_judge=to_user.user_id, 398 timestamp=datetime.now(), 399 question=question, 400 context=context, 401 pattern_signature=self._compute_pattern_signature(question, context), 402 reason=reason, 403 confidence_score=confidence 404 ) 405 406 self.pending_escalations.append(escalation) 407 from_user.escalations_made += 1 408 409 return escalation 410 411 def resolve_escalation( 412 self, 413 escalation_id: str, 414 decision: str, 415 reasoning: str, 416 judge: UserCognitiveState, 417 create_precedent: bool = True 418 ) -> Optional[Precedent]: 419 """ 420 Resolve an escalation with a judgment. 421 422 If create_precedent is True, this becomes tribal law. 423 """ 424 escalation = next( 425 (e for e in self.pending_escalations if e.id == escalation_id), 426 None 427 ) 428 if not escalation: 429 return None 430 431 escalation.resolved = True 432 escalation.resolution = decision 433 434 self.pending_escalations.remove(escalation) 435 self.resolved_escalations.append(escalation) 436 437 if create_precedent: 438 precedent = Precedent( 439 id=f"precedent_{escalation.id}", 440 judge_id=judge.user_id, 441 judge_role=judge.role, 442 judgment_date=datetime.now(), 443 original_question=escalation.question, 444 context=escalation.context, 445 pattern_signature=escalation.pattern_signature, 446 decision=decision, 447 reasoning=reasoning, 448 applies_to_patterns=[escalation.pattern_signature] 449 ) 450 451 self.collective.add_precedent(precedent) 452 judge.precedents_set += 1 453 escalation.became_precedent = True 454 455 return precedent 456 457 return None 458 459 def _compute_pattern_signature(self, question: str, context: str) -> str: 460 """Compute a pattern signature for matching.""" 461 # Extract key concepts 462 combined = f"{question} {context}".lower() 463 words = combined.split() 464 # Simple approach: sorted unique words 465 key_words = sorted(set(w for w in words if len(w) > 3))[:10] 466 return " ".join(key_words) 467 468 def _find_conflicting_wells(self, question: str, user: UserCognitiveState) -> List[str]: 469 """Find gravity wells that might conflict for this question.""" 470 question_lower = question.lower() 471 relevant_wells = [] 472 for concept, well in user.gravity_wells.items(): 473 if concept.lower() in question_lower: 474 relevant_wells.append(concept) 475 return relevant_wells 476 477 def _generate_id(self, question: str) -> str: 478 """Generate unique ID for escalation.""" 479 content = f"{question}_{datetime.now().isoformat()}" 480 return hashlib.md5(content.encode()).hexdigest()[:12] 481 482 483 class CampfireManager: 484 """ 485 The campfire is where individual and collective sense-making meet. 486 """ 487 488 def __init__(self, collective: CollectiveGraph): 489 self.collective = collective 490 491 def run_campfire( 492 self, 493 users: List[UserCognitiveState], 494 scope: str = "full" # "full", "team", "daily" 495 ) -> CampfireReport: 496 """ 497 Run a campfire synchronization. 498 499 Individual models inform collective, collective informs individuals. 500 """ 501 timestamp = datetime.now() 502 participants = [u.user_id for u in users] 503 504 # 1. GATHER: Collect individual signals 505 individual_signals = self._gather_individual_signals(users) 506 507 # 2. WEIGHT: Weight by tenure and role 508 weighted_signals = self._weight_by_tenure(individual_signals, users) 509 510 # 3. UPDATE COLLECTIVE: New patterns emerge 511 new_wells, strengthened, decayed = self._update_collective(weighted_signals, users) 512 513 # 4. BROADCAST: Collective insights flow to individuals 514 individual_updates = self._broadcast_to_individuals(users) 515 516 # 5. CALIBRATE: Individuals adjust based on collective 517 self._calibrate_individuals(users, individual_updates) 518 519 # Record campfire 520 report = CampfireReport( 521 timestamp=timestamp, 522 participants=participants, 523 new_collective_wells=new_wells, 524 strengthened_wells=strengthened, 525 decayed_wells=decayed, 526 individual_updates=individual_updates, 527 new_precedents=[], 528 escalations=[] 529 ) 530 531 self.collective.campfire_history.append({ 532 'timestamp': timestamp.isoformat(), 533 'participants': participants, 534 'scope': scope, 535 'new_wells': len(new_wells), 536 'strengthened': len(strengthened) 537 }) 538 539 # Update user campfire timestamps 540 for user in users: 541 user.last_campfire = timestamp 542 543 return report 544 545 def _gather_individual_signals(self, users: List[UserCognitiveState]) -> Dict[str, Dict]: 546 """What has each person learned since last campfire?""" 547 signals = {} 548 for user in users: 549 # Get recent gravity well activations 550 recent_wells = [ 551 (concept, well) 552 for concept, well in user.gravity_wells.items() 553 if user.last_campfire is None or well.last_activated > user.last_campfire 554 ] 555 556 signals[user.user_id] = { 557 'recent_wells': recent_wells, 558 'pattern_preferences': user.pattern_preferences, 559 'altitude_distribution': user.altitude_distribution, 560 'tenure_days': user.tenure_days, 561 'role': user.role 562 } 563 564 return signals 565 566 def _weight_by_tenure( 567 self, 568 signals: Dict[str, Dict], 569 users: List[UserCognitiveState] 570 ) -> Dict[str, float]: 571 """Weight contributions by tenure and role.""" 572 weights = {} 573 574 for user in users: 575 signal = signals.get(user.user_id, {}) 576 577 # Base weight by tenure (logarithmic, not linear) 578 tenure_weight = min(1.0, math.log1p(user.tenure_days / 30) / 4) 579 580 # Role modifier 581 role_modifiers = { 582 TribalRole.FOUNDER: 2.0, 583 TribalRole.ELDER: 1.5, 584 TribalRole.WORKING: 1.0, 585 TribalRole.YOUNG: 0.7 586 } 587 role_mod = role_modifiers.get(user.role, 1.0) 588 589 # Fresh perspective bonus for young 590 if user.role == TribalRole.YOUNG: 591 # Young users with active wells get bonus 592 if signal.get('recent_wells'): 593 role_mod += 0.3 594 595 weights[user.user_id] = tenure_weight * role_mod 596 597 return weights 598 599 def _update_collective( 600 self, 601 weights: Dict[str, float], 602 users: List[UserCognitiveState] 603 ) -> Tuple[List[str], List[str], List[str]]: 604 """Update collective graph based on weighted individual signals.""" 605 new_wells = [] 606 strengthened = [] 607 decayed = [] 608 609 # Aggregate gravity well votes 610 well_votes: Dict[str, List[Tuple[float, str]]] = defaultdict(list) 611 612 for user in users: 613 weight = weights.get(user.user_id, 1.0) 614 for concept, well in user.gravity_wells.items(): 615 well_votes[concept].append((well.mass * weight, user.user_id)) 616 617 # Process votes 618 for concept, votes in well_votes.items(): 619 total_weight = sum(v[0] for v in votes) 620 contributor_count = len(votes) 621 contributors = [v[1] for v in votes] 622 623 # Multiple people care about this = collective well 624 if contributor_count >= 2 or total_weight > 0.5: 625 if concept not in self.collective.collective_wells: 626 # New collective well 627 self.collective.collective_wells[concept] = GravityWell( 628 concept=concept, 629 mass=total_weight / contributor_count, 630 contributors=contributors 631 ) 632 new_wells.append(concept) 633 else: 634 # Strengthen existing 635 existing = self.collective.collective_wells[concept] 636 existing.mass = min(0.95, existing.mass + 0.1) 637 existing.activation_count += 1 638 existing.last_activated = datetime.now() 639 for c in contributors: 640 if c not in existing.contributors: 641 existing.contributors.append(c) 642 strengthened.append(concept) 643 644 # Decay wells that weren't activated 645 for concept, well in list(self.collective.collective_wells.items()): 646 if concept not in well_votes: 647 days_since = (datetime.now() - well.last_activated).days 648 if days_since > 0: 649 well.decay(days_since) 650 if well.mass < 0.15: 651 decayed.append(concept) 652 653 # Remove very weak wells 654 for concept in decayed: 655 if self.collective.collective_wells[concept].mass < 0.1: 656 del self.collective.collective_wells[concept] 657 658 self.collective.updated_at = datetime.now() 659 660 return new_wells, strengthened, decayed 661 662 def _broadcast_to_individuals(self, users: List[UserCognitiveState]) -> Dict[str, Dict]: 663 """Collective insights flow back to individuals.""" 664 updates = {} 665 666 for user in users: 667 user_update = { 668 'suggested_wells': [], 669 'collective_stronger': [], 670 'new_precedents': [] 671 } 672 673 # What collective wells should this user know about? 674 for concept, well in self.collective.collective_wells.items(): 675 if concept not in user.gravity_wells: 676 # User doesn't have this well - suggest it 677 if well.mass > 0.5: 678 user_update['suggested_wells'].append({ 679 'concept': concept, 680 'collective_mass': well.mass, 681 'contributors': well.contributors[:3] 682 }) 683 else: 684 # Compare strengths 685 user_strength = user.gravity_wells[concept].mass 686 if user_strength < well.mass * 0.5: 687 user_update['collective_stronger'].append({ 688 'concept': concept, 689 'user_mass': user_strength, 690 'collective_mass': well.mass 691 }) 692 693 updates[user.user_id] = user_update 694 695 return updates 696 697 def _calibrate_individuals(self, users: List[UserCognitiveState], updates: Dict[str, Dict]): 698 """Individuals adjust based on collective.""" 699 for user in users: 700 user_update = updates.get(user.user_id, {}) 701 702 # Adjust collective_weight based on alignment 703 alignment = self._compute_alignment(user) 704 # Users naturally drift toward collective over time 705 if alignment > 0.7: 706 user.collective_weight = min(0.8, user.collective_weight + 0.05) 707 elif alignment < 0.3: 708 user.collective_weight = max(0.2, user.collective_weight - 0.05) 709 710 def _compute_alignment(self, user: UserCognitiveState) -> float: 711 """How well does user align with collective?""" 712 if not user.gravity_wells or not self.collective.collective_wells: 713 return 0.5 714 715 user_concepts = set(user.gravity_wells.keys()) 716 collective_concepts = set(self.collective.collective_wells.keys()) 717 718 if not user_concepts: 719 return 0.5 720 721 overlap = user_concepts & collective_concepts 722 return len(overlap) / len(user_concepts) 723 724 725 class TribalCognition: 726 """ 727 Main interface for tribal cognition system. 728 729 Integrates with First Officer to add multi-user support. 730 """ 731 732 def __init__(self, founder_id: str = "founder"): 733 self.founder_id = founder_id 734 self.users: Dict[str, UserCognitiveState] = {} 735 self.collective = CollectiveGraph() 736 self.role_assigner = RoleAssigner() 737 self.escalation_manager = EscalationManager(self.collective) 738 self.campfire_manager = CampfireManager(self.collective) 739 740 # Callbacks 741 self._on_precedent: List[Callable[[Precedent], None]] = [] 742 self._on_escalation: List[Callable[[EscalationRequest], None]] = [] 743 744 def register_user( 745 self, 746 user_id: str, 747 display_name: str, 748 join_date: datetime = None, 749 is_founder: bool = False 750 ) -> UserCognitiveState: 751 """Register a new user.""" 752 user = UserCognitiveState( 753 user_id=user_id, 754 display_name=display_name, 755 join_date=join_date or datetime.now(), 756 role=TribalRole.FOUNDER if is_founder else TribalRole.YOUNG 757 ) 758 759 # Assign role 760 user.role = self.role_assigner.assign_role(user, self.collective, is_founder) 761 762 self.users[user_id] = user 763 return user 764 765 def get_user(self, user_id: str) -> Optional[UserCognitiveState]: 766 """Get user by ID.""" 767 return self.users.get(user_id) 768 769 def record_user_activity( 770 self, 771 user_id: str, 772 topics: List[str], 773 altitude: str = None, 774 content: str = None 775 ): 776 """Record activity for a user, updating their gravity wells.""" 777 user = self.users.get(user_id) 778 if not user: 779 return 780 781 user.last_active = datetime.now() 782 783 # Update gravity wells 784 for topic in topics: 785 user.add_gravity_well(topic) 786 787 # Update altitude distribution 788 if altitude: 789 total = sum(user.altitude_distribution.values()) 790 for alt in user.altitude_distribution: 791 if alt == altitude: 792 user.altitude_distribution[alt] = min(0.5, user.altitude_distribution[alt] + 0.02) 793 else: 794 user.altitude_distribution[alt] = max(0.1, user.altitude_distribution[alt] - 0.01) 795 # Normalize 796 new_total = sum(user.altitude_distribution.values()) 797 for alt in user.altitude_distribution: 798 user.altitude_distribution[alt] /= new_total 799 800 def check_escalation( 801 self, 802 user_id: str, 803 question: str, 804 context: str, 805 confidence: float, 806 stakes: str = "medium" 807 ) -> Tuple[bool, str, Optional[Precedent]]: 808 """ 809 Check if a decision should be escalated. 810 811 Returns: (should_escalate, reason, existing_precedent_if_any) 812 """ 813 user = self.users.get(user_id) 814 if not user: 815 return False, "User not found", None 816 817 # Check for existing precedent first 818 pattern = self.escalation_manager._compute_pattern_signature(question, context) 819 precedent = self.collective.find_precedent(pattern) 820 if precedent: 821 precedent.apply() 822 return False, f"Apply precedent: {precedent.decision}", precedent 823 824 # Check if should escalate 825 should, reason = self.escalation_manager.should_escalate( 826 question, context, user, confidence, stakes 827 ) 828 829 return should, reason, None 830 831 def escalate( 832 self, 833 from_user_id: str, 834 question: str, 835 context: str, 836 confidence: float, 837 reason: str 838 ) -> Optional[EscalationRequest]: 839 """Create an escalation to an appropriate elder.""" 840 from_user = self.users.get(from_user_id) 841 if not from_user: 842 return None 843 844 # Find appropriate judge 845 to_user = self._find_judge(from_user) 846 if not to_user: 847 return None 848 849 escalation = self.escalation_manager.create_escalation( 850 question=question, 851 context=context, 852 from_user=from_user, 853 to_user=to_user, 854 reason=reason, 855 confidence=confidence 856 ) 857 858 # Notify callbacks 859 for callback in self._on_escalation: 860 try: 861 callback(escalation) 862 except Exception: 863 pass 864 865 return escalation 866 867 def resolve_escalation( 868 self, 869 escalation_id: str, 870 judge_id: str, 871 decision: str, 872 reasoning: str, 873 create_precedent: bool = True 874 ) -> Optional[Precedent]: 875 """Judge resolves an escalation.""" 876 judge = self.users.get(judge_id) 877 if not judge: 878 return None 879 880 precedent = self.escalation_manager.resolve_escalation( 881 escalation_id=escalation_id, 882 decision=decision, 883 reasoning=reasoning, 884 judge=judge, 885 create_precedent=create_precedent 886 ) 887 888 if precedent: 889 # Notify callbacks 890 for callback in self._on_precedent: 891 try: 892 callback(precedent) 893 except Exception: 894 pass 895 896 # Notify all users 897 for user in self.users.values(): 898 user.precedents_received += 1 899 900 return precedent 901 902 def run_campfire(self, user_ids: List[str] = None, scope: str = "full") -> CampfireReport: 903 """Run a campfire synchronization.""" 904 if user_ids: 905 users = [self.users[uid] for uid in user_ids if uid in self.users] 906 else: 907 users = list(self.users.values()) 908 909 return self.campfire_manager.run_campfire(users, scope) 910 911 def _find_judge(self, from_user: UserCognitiveState) -> Optional[UserCognitiveState]: 912 """Find appropriate judge for escalation.""" 913 # Find users with higher role 914 role_order = [TribalRole.YOUNG, TribalRole.WORKING, TribalRole.ELDER, TribalRole.FOUNDER] 915 from_role_idx = role_order.index(from_user.role) 916 917 candidates = [ 918 u for u in self.users.values() 919 if role_order.index(u.role) > from_role_idx 920 ] 921 922 if not candidates: 923 # If no higher role, find most senior peer 924 candidates = [ 925 u for u in self.users.values() 926 if u.user_id != from_user.user_id and u.tenure_days > from_user.tenure_days 927 ] 928 929 if not candidates: 930 return None 931 932 # Prefer founder for serious matters 933 founder = next((u for u in candidates if u.role == TribalRole.FOUNDER), None) 934 if founder: 935 return founder 936 937 # Otherwise most senior elder 938 candidates.sort(key=lambda u: (-role_order.index(u.role), -u.tenure_days)) 939 return candidates[0] 940 941 def get_context_for_user(self, user_id: str) -> str: 942 """Get tribal context to inject into user's session.""" 943 user = self.users.get(user_id) 944 if not user: 945 return "" 946 947 lines = ["<tribal-context>"] 948 949 # User's role 950 lines.append(f"Your role: {user.role.value} (tenure: {user.tenure_days} days)") 951 952 # Collective gravity wells 953 if self.collective.collective_wells: 954 lines.append("") 955 lines.append("Collective priorities (what the tribe cares about):") 956 top_wells = sorted( 957 self.collective.collective_wells.items(), 958 key=lambda x: -x[1].mass 959 )[:5] 960 for concept, well in top_wells: 961 your_strength = user.get_gravity_for_concept(concept) 962 alignment = "aligned" if your_strength > 0.3 else "new to you" 963 lines.append(f" - {concept} (mass: {well.mass:.2f}, {alignment})") 964 965 # Recent precedents 966 recent_precedents = [ 967 p for p in self.collective.precedents[-5:] 968 if p.times_applied < 3 # Still relatively new 969 ] 970 if recent_precedents: 971 lines.append("") 972 lines.append("Recent precedents (apply these locally):") 973 for p in recent_precedents: 974 lines.append(f" - {p.decision[:50]}... (from {p.judge_role.value})") 975 976 # Pending escalations to this user 977 pending = [ 978 e for e in self.escalation_manager.pending_escalations 979 if e.to_judge == user_id 980 ] 981 if pending: 982 lines.append("") 983 lines.append(f"Pending escalations awaiting your judgment: {len(pending)}") 984 for e in pending[:2]: 985 lines.append(f" - From {e.from_user}: {e.question[:50]}...") 986 987 lines.append("</tribal-context>") 988 return "\n".join(lines) 989 990 def on_precedent(self, callback: Callable[[Precedent], None]): 991 """Register callback for new precedents.""" 992 self._on_precedent.append(callback) 993 994 def on_escalation(self, callback: Callable[[EscalationRequest], None]): 995 """Register callback for new escalations.""" 996 self._on_escalation.append(callback) 997 998 def get_tribe_state(self) -> Dict[str, Any]: 999 """Get state for dashboard visualization.""" 1000 return { 1001 'users': { 1002 uid: { 1003 'display_name': u.display_name, 1004 'role': u.role.value, 1005 'tenure_days': u.tenure_days, 1006 'gravity_wells': len(u.gravity_wells), 1007 'collective_weight': u.collective_weight, 1008 'last_active': u.last_active.isoformat() if u.last_active else None 1009 } 1010 for uid, u in self.users.items() 1011 }, 1012 'collective': { 1013 'wells': { 1014 concept: { 1015 'mass': well.mass, 1016 'contributors': well.contributors[:3] 1017 } 1018 for concept, well in self.collective.collective_wells.items() 1019 }, 1020 'precedent_count': len(self.collective.precedents), 1021 'campfire_count': len(self.collective.campfire_history) 1022 }, 1023 'pending_escalations': len(self.escalation_manager.pending_escalations), 1024 'updated_at': self.collective.updated_at.isoformat() 1025 } 1026 1027 1028 # Singleton instance 1029 _tribal_cognition: Optional[TribalCognition] = None 1030 1031 1032 def get_tribal_cognition() -> TribalCognition: 1033 """Get or create the global tribal cognition instance.""" 1034 global _tribal_cognition 1035 if _tribal_cognition is None: 1036 _tribal_cognition = TribalCognition() 1037 return _tribal_cognition