resonance.py
1 """ 2 Resonance Detector for Sovereign OS Mission Control 3 4 Detects shared patterns across multiple threads and calculates 5 resonance strength for determining when to alert. 6 7 Resonance Types: 8 - CONVERGENCE: Threads reaching similar conclusions 9 - DIVERGENCE: Threads contradicting each other 10 - SHARED_CONCEPT: Same idea in multiple threads 11 - DEPENDENCY: Thread A needs Thread B to complete 12 13 Typed Resonance (from typed-resonance-principle.md): 14 - Every resonance carries semantic type information 15 - Connections know WHY they exist (which axiom principle) 16 - Generic similarity hides; typed resonance reveals 17 """ 18 19 import re 20 import json 21 import logging 22 import urllib.request 23 import urllib.error 24 from pathlib import Path 25 from dataclasses import dataclass, field 26 from typing import Dict, List, Set, Optional, Tuple, Any 27 from datetime import datetime 28 from enum import Enum 29 from collections import Counter 30 31 logger = logging.getLogger(__name__) 32 33 # Mesh network integration (N of X - all instances share resonance state) 34 MESH_HTTP_PORT = 7778 35 36 37 def publish_to_mesh(message_type: str, payload: Dict[str, Any]) -> bool: 38 """Publish a resonance event to the sovereign mesh network. 39 40 Args: 41 message_type: Event type (resonance_detected, resonance_alert, etc.) 42 payload: Event data to broadcast 43 44 Returns: 45 True if published successfully, False otherwise 46 """ 47 try: 48 msg = json.dumps({ 49 "type": message_type, 50 "from": "resonance-detector", 51 "payload": payload, 52 "timestamp": datetime.now().isoformat() 53 }).encode('utf-8') 54 55 req = urllib.request.Request( 56 f"http://localhost:{MESH_HTTP_PORT}/publish", 57 data=msg, 58 headers={"Content-Type": "application/json"}, 59 method="POST" 60 ) 61 with urllib.request.urlopen(req, timeout=2) as resp: 62 return resp.status == 200 63 except (urllib.error.URLError, TimeoutError, ConnectionRefusedError): 64 # Mesh not available - not critical, continue operation 65 return False 66 except Exception: 67 return False 68 69 70 # ============================================================================= 71 # TYPED RESONANCE - Axiom Semantic Fields 72 # From typed-resonance-principle.md: "Resonance is typed, not generic" 73 # ============================================================================= 74 75 AXIOM_FIELDS = { 76 "A0": { 77 "name": "Boundary Operation", 78 "strong": {'boundary', 'blanket', 'markov', 'distinction', 'observer', 79 'sovereign', 'sovereignty', 'sensory', 'active', 'internal', 'external'}, 80 "medium": {'permeable', 'permeability', 'flow', 'flows', 'structure', 'content', 81 'scale', 'fractal', 'nested', 'inside', 'outside', 'crosses'}, 82 "weak": {'separate', 'division', 'filter', 'membrane'}, 83 }, 84 "A1": { 85 "name": "Telos of Integration", 86 "strong": {'integration', 'integrate', 'integrating', 'connection', 'connect', 87 'binding', 'isolation', 'isolated', 'isolating', 'relation'}, 88 "medium": {'tribe', 'tribal', 'collective', 'unify', 'unity', 'unified', 89 'merge', 'merging', 'together', 'shared'}, 90 "weak": {'join', 'joining', 'link', 'linking', 'collaborate'}, 91 }, 92 "A2": { 93 "name": "Recognition of Life", 94 "strong": {'life', 'alive', 'living', 'death', 'dead', 'dying', 'primitive', 95 'calcified', 'sclerosis', 'ornament', 'cain', 'blindness'}, 96 "medium": {'recognize', 'recognition', 'beauty', 'beautiful', 'resonance', 97 'simple', 'complex', 'accumulated', 'cruft', 'golden', 'carpenter'}, 98 "weak": {'fresh', 'stale', 'responsive', 'rigid', 'clear', 'opaque'}, 99 }, 100 "A3": { 101 "name": "Dynamic Pole Navigation", 102 "strong": {'pole', 'poles', 'dyad', 'tension', 'navigate', 'navigation', 103 'oscillate', 'oscillation', 'movement', 'shadow', 'dynamic', 'static'}, 104 "medium": {'extreme', 'extremes', 'balance', 'between', 'spectrum', 105 'context', 'contextual', 'invert', 'inversion'}, 106 "weak": {'middle', 'midpoint', 'swing', 'shift', 'pendulum'}, 107 }, 108 "A4": { 109 "name": "Ergodic Asymmetry", 110 "strong": {'ruin', 'ruinous', 'catastrophe', 'catastrophic', 'terminal', 111 'irreversible', 'unrecoverable', 'survival', 'survive', 'ergodic', 112 'asymmetry', 'asymmetric', 'compound'}, 113 "medium": {'risk', 'risky', 'dangerous', 'cheap', 'expensive', 'cost', 114 'rebuild', 'rewrite', 'redo', 'reversible', 'undo', 'rollback'}, 115 "weak": {'careful', 'caution', 'fail', 'failure', 'recover', 'backup'}, 116 }, 117 } 118 119 120 def calculate_axiom_resonance(terms: Set[str], axiom_id: str = None) -> Dict[str, float]: 121 """ 122 Calculate how strongly a set of terms resonates with each axiom. 123 124 Args: 125 terms: Set of terms (words) to analyze 126 axiom_id: If provided, only calculate for this axiom 127 128 Returns: 129 Dict mapping axiom ID to resonance score (0-1) 130 """ 131 word_set = {t.lower() for t in terms} 132 results = {} 133 134 axioms_to_check = [axiom_id] if axiom_id else AXIOM_FIELDS.keys() 135 136 for aid in axioms_to_check: 137 if aid not in AXIOM_FIELDS: 138 continue 139 axiom_field = AXIOM_FIELDS[aid] 140 141 strong_matches = word_set & axiom_field["strong"] 142 medium_matches = word_set & axiom_field["medium"] 143 weak_matches = word_set & axiom_field["weak"] 144 145 # Weighted score: strong=0.15, medium=0.08, weak=0.03 per term 146 score = ( 147 len(strong_matches) * 0.15 + 148 len(medium_matches) * 0.08 + 149 len(weak_matches) * 0.03 150 ) 151 results[aid] = min(1.0, score) 152 153 return results 154 155 156 def get_dominant_axiom(terms: Set[str]) -> Optional[Tuple[str, float]]: 157 """Get the dominant axiom for a set of terms, if any meets threshold.""" 158 scores = calculate_axiom_resonance(terms) 159 if not scores: 160 return None 161 162 dominant = max(scores.items(), key=lambda x: x[1]) 163 if dominant[1] >= 0.10: # Minimum threshold 164 return dominant 165 return None 166 167 168 def get_shared_axiom_resonance( 169 terms1: Set[str], 170 terms2: Set[str], 171 min_threshold: float = 0.10 172 ) -> Optional[Tuple[str, float]]: 173 """ 174 Find if two term sets share resonance with the same axiom. 175 176 Returns the shared axiom with highest combined resonance, or None. 177 """ 178 scores1 = calculate_axiom_resonance(terms1) 179 scores2 = calculate_axiom_resonance(terms2) 180 181 shared = {} 182 for axiom_id in scores1: 183 if axiom_id in scores2: 184 # Both must meet minimum threshold 185 if scores1[axiom_id] >= min_threshold and scores2[axiom_id] >= min_threshold: 186 shared[axiom_id] = min(scores1[axiom_id], scores2[axiom_id]) 187 188 if not shared: 189 return None 190 191 # Return highest shared resonance 192 best = max(shared.items(), key=lambda x: x[1]) 193 return best 194 195 196 class ResonanceType(Enum): 197 """Types of cross-thread resonance.""" 198 CONVERGENCE = "convergence" 199 DIVERGENCE = "divergence" 200 SHARED_CONCEPT = "shared_concept" 201 DEPENDENCY = "dependency" 202 203 204 @dataclass 205 class Resonance: 206 """A detected resonance across threads.""" 207 208 type: ResonanceType 209 pattern: str 210 threads: List[str] 211 strength: float # 0-1 212 timestamp: datetime 213 description: str = "" 214 215 # Typed resonance fields 216 axiom_type: Optional[str] = None # e.g., "A0", "A1", etc. 217 axiom_confidence: float = 0.0 # How strongly typed (0-1) 218 219 @property 220 def axiom_name(self) -> Optional[str]: 221 """Get the human-readable axiom name.""" 222 if self.axiom_type and self.axiom_type in AXIOM_FIELDS: 223 return AXIOM_FIELDS[self.axiom_type]["name"] 224 return None 225 226 @property 227 def urgency(self) -> float: 228 """ 229 Calculate resonance urgency. 230 231 Resonance_U = (Strength × 0.4) + (Thread_Count × 0.3) + (Blocking × 0.3) 232 """ 233 thread_factor = min(len(self.threads) / 4, 1.0) # Normalize to 4 threads 234 blocking = 1.0 if self.type == ResonanceType.DEPENDENCY else 0.0 235 236 return (self.strength * 0.4) + (thread_factor * 0.3) + (blocking * 0.3) 237 238 @property 239 def should_bubble(self) -> bool: 240 """Check if this resonance should bubble to main thread.""" 241 return self.urgency >= 0.7 242 243 @property 244 def should_highlight(self) -> bool: 245 """Check if this resonance should be highlighted in sidebar.""" 246 return 0.4 <= self.urgency < 0.7 247 248 def to_alert_markdown(self) -> str: 249 """Generate markdown for a RESONANCE-ALERT file.""" 250 # Axiom type line if present 251 axiom_line = "" 252 if self.axiom_type: 253 axiom_line = f"**Axiom Type:** {self.axiom_type} ({self.axiom_name}) - {self.axiom_confidence:.0%} confidence\n" 254 255 return f"""# Resonance Alert - {self.timestamp.strftime('%Y-%m-%d %H:%M')} 256 257 **Type:** {self.type.value.upper()} 258 {axiom_line}**Detected:** {self.timestamp.isoformat()} 259 **Affected Threads:** {', '.join(self.threads)} 260 **Strength:** {self.strength:.2f} 261 **Urgency:** {self.urgency:.2f} 262 263 --- 264 265 ## What Was Detected 266 267 {self.description} 268 269 **Pattern:** `{self.pattern}` 270 {self._generate_axiom_context()} 271 272 ## Affected Threads 273 274 {self._generate_thread_sections()} 275 276 ## Integration Recommendation 277 278 {self._generate_recommendation()} 279 280 ## Priority 281 282 **{'HIGH' if self.should_bubble else 'MEDIUM' if self.should_highlight else 'LOW'}** - {self._priority_rationale()} 283 284 --- 285 286 *Generated by Mission Control | {self.timestamp.strftime('%Y-%m-%d %H:%M')}* 287 """ 288 289 def _generate_axiom_context(self) -> str: 290 """Generate axiom context section if typed.""" 291 if not self.axiom_type: 292 return "" 293 294 return f""" 295 ### Why This Matters ({self.axiom_type}) 296 297 This resonance is typed as **{self.axiom_name}**. Both threads share semantic 298 resonance with this principle, suggesting a deeper conceptual connection 299 beyond surface-level term overlap. 300 301 **Typed resonance reveals meaning; generic resonance hides it.** 302 """ 303 304 def _generate_thread_sections(self) -> str: 305 sections = [] 306 for thread in self.threads: 307 sections.append(f"""### {thread} 308 - Pattern appears in this thread 309 - Suggested action: Review for alignment with other threads 310 """) 311 return "\n".join(sections) 312 313 def _generate_recommendation(self) -> str: 314 if self.type == ResonanceType.CONVERGENCE: 315 return "Consider creating a sync point or merging conclusions." 316 elif self.type == ResonanceType.DIVERGENCE: 317 return "Human resolution needed - threads have conflicting conclusions." 318 elif self.type == ResonanceType.DEPENDENCY: 319 return "Resolve blocking thread first to unblock dependent threads." 320 else: 321 return "Review shared concept for consistency across threads." 322 323 def _priority_rationale(self) -> str: 324 if self.should_bubble: 325 return f"Urgency {self.urgency:.2f} exceeds bubble threshold (0.7)" 326 elif self.should_highlight: 327 return f"Urgency {self.urgency:.2f} warrants sidebar highlight" 328 else: 329 return f"Urgency {self.urgency:.2f} is low priority" 330 331 332 @dataclass 333 class ThreadSnapshot: 334 """Snapshot of a thread's state for resonance analysis.""" 335 336 thread_id: str 337 patterns: Set[str] 338 conclusions: List[str] 339 dependencies: List[str] 340 gravity_wells: Dict[str, float] 341 last_updated: datetime 342 343 # Typed resonance fields 344 axiom_resonance: Dict[str, float] = field(default_factory=dict) # axiom_id -> score 345 dominant_axiom: Optional[str] = None 346 347 def update_axiom_resonance(self) -> None: 348 """Recalculate axiom resonance from patterns.""" 349 self.axiom_resonance = calculate_axiom_resonance(self.patterns) 350 dominant = get_dominant_axiom(self.patterns) 351 self.dominant_axiom = dominant[0] if dominant else None 352 353 354 class ResonanceDetector: 355 """ 356 Detects resonance across multiple thread states. 357 358 Resonance = patterns that appear meaningfully across threads. 359 Not just shared words, but shared concepts with semantic weight. 360 """ 361 362 # Common words to ignore 363 STOP_WORDS = { 364 "the", "a", "an", "is", "are", "was", "were", "be", "been", 365 "being", "have", "has", "had", "do", "does", "did", "will", 366 "would", "could", "should", "may", "might", "must", "shall", 367 "can", "this", "that", "these", "those", "it", "its", 368 "and", "or", "but", "if", "then", "else", "when", "where", 369 "which", "who", "what", "how", "why", "all", "each", "every", 370 "both", "few", "more", "most", "other", "some", "such", "no", 371 "not", "only", "same", "so", "than", "too", "very", "just", 372 "also", "now", "here", "there", "for", "with", "about", "into" 373 } 374 375 # Minimum pattern length to consider 376 MIN_PATTERN_LENGTH = 3 377 378 # Minimum threads for resonance 379 MIN_THREADS_FOR_RESONANCE = 2 380 381 def __init__(self): 382 self._thread_snapshots: Dict[str, ThreadSnapshot] = {} 383 self._detected_resonances: List[Resonance] = [] 384 385 def update_thread(self, snapshot: ThreadSnapshot) -> None: 386 """Update the detector with a new thread snapshot.""" 387 # Calculate axiom resonance for the thread 388 snapshot.update_axiom_resonance() 389 self._thread_snapshots[snapshot.thread_id] = snapshot 390 391 def detect(self) -> List[Resonance]: 392 """ 393 Detect all resonances across current thread states. 394 395 Returns list of newly detected resonances. 396 """ 397 if len(self._thread_snapshots) < self.MIN_THREADS_FOR_RESONANCE: 398 return [] 399 400 resonances = [] 401 402 # Detect shared concepts (with axiom typing) 403 shared_concepts = self._find_shared_concepts() 404 for pattern, threads in shared_concepts.items(): 405 strength, axiom_type, axiom_conf = self._calculate_concept_strength(pattern, threads) 406 if strength > 0.3: # Threshold for meaningful resonance 407 resonances.append(Resonance( 408 type=ResonanceType.SHARED_CONCEPT, 409 pattern=pattern, 410 threads=list(threads), 411 strength=strength, 412 timestamp=datetime.now(), 413 description=f"Concept '{pattern}' appears in {len(threads)} threads", 414 axiom_type=axiom_type, 415 axiom_confidence=axiom_conf, 416 )) 417 418 # Detect typed convergence (threads sharing axiom resonance) 419 typed_convergences = self._find_typed_convergence() 420 resonances.extend(typed_convergences) 421 422 # Detect convergence (similar conclusions) 423 convergences = self._find_convergence() 424 resonances.extend(convergences) 425 426 # Detect divergence (conflicting conclusions) 427 divergences = self._find_divergence() 428 resonances.extend(divergences) 429 430 # Detect dependencies 431 dependencies = self._find_dependencies() 432 resonances.extend(dependencies) 433 434 # Publish to mesh (N of X - all instances see resonance events) 435 for resonance in resonances: 436 publish_to_mesh("resonance_detected", { 437 "type": resonance.type.value, 438 "pattern": resonance.pattern, 439 "threads": resonance.threads, 440 "strength": resonance.strength, 441 "urgency": resonance.urgency, 442 "axiom_type": resonance.axiom_type, 443 "axiom_confidence": resonance.axiom_confidence, 444 "should_bubble": resonance.should_bubble, 445 "description": resonance.description[:200] # Truncate for mesh 446 }) 447 448 # Store and return 449 self._detected_resonances.extend(resonances) 450 return resonances 451 452 def _find_shared_concepts(self) -> Dict[str, Set[str]]: 453 """Find patterns that appear in multiple threads.""" 454 pattern_threads: Dict[str, Set[str]] = {} 455 456 for thread_id, snapshot in self._thread_snapshots.items(): 457 for pattern in snapshot.patterns: 458 # Normalize pattern 459 normalized = self._normalize_pattern(pattern) 460 if not normalized: 461 continue 462 463 if normalized not in pattern_threads: 464 pattern_threads[normalized] = set() 465 pattern_threads[normalized].add(thread_id) 466 467 # Filter to patterns in 2+ threads 468 return { 469 p: t for p, t in pattern_threads.items() 470 if len(t) >= self.MIN_THREADS_FOR_RESONANCE 471 } 472 473 def _normalize_pattern(self, pattern: str) -> Optional[str]: 474 """Normalize a pattern for comparison.""" 475 # Lowercase 476 normalized = pattern.lower().strip() 477 478 # Skip if too short 479 if len(normalized) < self.MIN_PATTERN_LENGTH: 480 return None 481 482 # Skip stop words 483 if normalized in self.STOP_WORDS: 484 return None 485 486 # Remove common prefixes/suffixes 487 normalized = re.sub(r'^(the|a|an)\s+', '', normalized) 488 normalized = re.sub(r'\s+(protocol|pattern|concept)$', '', normalized) 489 490 return normalized if normalized else None 491 492 def _calculate_concept_strength( 493 self, 494 pattern: str, 495 threads: Set[str] 496 ) -> Tuple[float, Optional[str], float]: 497 """ 498 Calculate the strength of a shared concept with typed resonance. 499 500 Factors: 501 - Number of threads (more = stronger) 502 - Gravity well strength in each thread 503 - Specificity of the pattern 504 - BONUS: Shared axiom resonance across threads 505 506 Returns: 507 Tuple of (strength, axiom_type, axiom_confidence) 508 """ 509 # Base strength from thread count 510 thread_factor = min(len(threads) / len(self._thread_snapshots), 1.0) 511 512 # Gravity well factor 513 gravity_sum = 0.0 514 for thread_id in threads: 515 snapshot = self._thread_snapshots.get(thread_id) 516 if snapshot and pattern in snapshot.gravity_wells: 517 gravity_sum += snapshot.gravity_wells[pattern] 518 gravity_factor = min(gravity_sum / len(threads), 1.0) if threads else 0 519 520 # Specificity factor (longer patterns are more specific) 521 specificity = min(len(pattern) / 20, 1.0) 522 523 base_strength = (thread_factor * 0.35) + (gravity_factor * 0.35) + (specificity * 0.15) 524 525 # === TYPED RESONANCE: Check for shared axiom resonance === 526 axiom_type = None 527 axiom_confidence = 0.0 528 529 # Collect all patterns from threads sharing this concept 530 thread_list = list(threads) 531 if len(thread_list) >= 2: 532 # Get patterns from first two threads 533 snap1 = self._thread_snapshots.get(thread_list[0]) 534 snap2 = self._thread_snapshots.get(thread_list[1]) 535 536 if snap1 and snap2: 537 shared = get_shared_axiom_resonance(snap1.patterns, snap2.patterns) 538 if shared: 539 axiom_type = shared[0] 540 axiom_confidence = shared[1] 541 # Boost strength for typed resonance (15% boost) 542 base_strength += 0.15 * axiom_confidence 543 544 return (min(base_strength, 1.0), axiom_type, axiom_confidence) 545 546 def _find_convergence(self) -> List[Resonance]: 547 """Find threads converging on similar conclusions.""" 548 # This would need more sophisticated NLP 549 # For now, look for similar conclusion patterns 550 resonances = [] 551 552 conclusion_patterns: Dict[str, List[Tuple[str, str]]] = {} 553 554 for thread_id, snapshot in self._thread_snapshots.items(): 555 for conclusion in snapshot.conclusions: 556 # Extract key terms from conclusion 557 key = self._extract_conclusion_key(conclusion) 558 if key: 559 if key not in conclusion_patterns: 560 conclusion_patterns[key] = [] 561 conclusion_patterns[key].append((thread_id, conclusion)) 562 563 for key, items in conclusion_patterns.items(): 564 if len(items) >= 2: 565 threads = [item[0] for item in items] 566 resonances.append(Resonance( 567 type=ResonanceType.CONVERGENCE, 568 pattern=key, 569 threads=threads, 570 strength=0.7, 571 timestamp=datetime.now(), 572 description=f"Threads converging on conclusion: {key}" 573 )) 574 575 return resonances 576 577 def _extract_conclusion_key(self, conclusion: str) -> Optional[str]: 578 """Extract a key pattern from a conclusion for comparison.""" 579 # Simple approach: first significant noun phrase 580 words = conclusion.lower().split() 581 significant = [w for w in words if w not in self.STOP_WORDS and len(w) > 3] 582 return " ".join(significant[:3]) if significant else None 583 584 def _find_divergence(self) -> List[Resonance]: 585 """Find threads with contradicting conclusions.""" 586 # Would need NLP for contradiction detection 587 # Placeholder for now 588 return [] 589 590 def _find_dependencies(self) -> List[Resonance]: 591 """Find thread dependencies.""" 592 resonances = [] 593 594 for thread_id, snapshot in self._thread_snapshots.items(): 595 for dep in snapshot.dependencies: 596 if dep in self._thread_snapshots: 597 resonances.append(Resonance( 598 type=ResonanceType.DEPENDENCY, 599 pattern=f"{thread_id} -> {dep}", 600 threads=[thread_id, dep], 601 strength=0.8, 602 timestamp=datetime.now(), 603 description=f"Thread '{thread_id}' depends on '{dep}'" 604 )) 605 606 return resonances 607 608 def _find_typed_convergence(self) -> List[Resonance]: 609 """ 610 Find threads that share axiom resonance (typed convergence). 611 612 This detects when multiple threads resonate with the same principle, 613 even if they don't share specific patterns. This is a DEEPER form 614 of convergence than shared concepts. 615 """ 616 resonances = [] 617 618 # Group threads by dominant axiom 619 axiom_threads: Dict[str, List[str]] = {} 620 for thread_id, snapshot in self._thread_snapshots.items(): 621 if snapshot.dominant_axiom: 622 if snapshot.dominant_axiom not in axiom_threads: 623 axiom_threads[snapshot.dominant_axiom] = [] 624 axiom_threads[snapshot.dominant_axiom].append(thread_id) 625 626 # Create convergence for axioms with 2+ threads 627 for axiom_id, threads in axiom_threads.items(): 628 if len(threads) >= 2: 629 axiom_name = AXIOM_FIELDS[axiom_id]["name"] 630 631 # Calculate average resonance strength 632 total_resonance = 0.0 633 for tid in threads: 634 snap = self._thread_snapshots.get(tid) 635 if snap and axiom_id in snap.axiom_resonance: 636 total_resonance += snap.axiom_resonance[axiom_id] 637 avg_resonance = total_resonance / len(threads) 638 639 resonances.append(Resonance( 640 type=ResonanceType.CONVERGENCE, 641 pattern=f"{axiom_id}: {axiom_name}", 642 threads=threads, 643 strength=0.6 + (0.3 * avg_resonance), # Base 0.6, scaled by resonance 644 timestamp=datetime.now(), 645 description=( 646 f"Threads converging on {axiom_id} ({axiom_name}). " 647 f"All {len(threads)} threads resonate with this principle." 648 ), 649 axiom_type=axiom_id, 650 axiom_confidence=avg_resonance, 651 )) 652 653 return resonances 654 655 def get_all_resonances(self) -> List[Resonance]: 656 """Get all detected resonances.""" 657 return self._detected_resonances.copy() 658 659 def get_high_urgency_resonances(self, threshold: float = 0.7) -> List[Resonance]: 660 """Get resonances above urgency threshold.""" 661 return [r for r in self._detected_resonances if r.urgency >= threshold] 662 663 def clear(self) -> None: 664 """Clear all state.""" 665 self._thread_snapshots.clear() 666 self._detected_resonances.clear() 667 668 669 # CLI for testing 670 if __name__ == "__main__": 671 print("=== Resonance Detector Test (with Typed Resonance) ===\n") 672 673 detector = ResonanceDetector() 674 675 # Test with axiom-aligned data 676 # Thread 1: A0 (Boundary) focused 677 detector.update_thread(ThreadSnapshot( 678 thread_id="architecture-review", 679 patterns={"boundary", "markov", "blanket", "permeability", "structure"}, 680 conclusions=["System needs clear boundary definitions"], 681 dependencies=[], 682 gravity_wells={"boundary": 0.9, "architecture": 0.7}, 683 last_updated=datetime.now() 684 )) 685 686 # Thread 2: Also A0 (Boundary) focused - should trigger typed convergence 687 detector.update_thread(ThreadSnapshot( 688 thread_id="security-protocol", 689 patterns={"boundary", "sovereign", "internal", "external", "flow"}, 690 conclusions=["Security is about boundary management"], 691 dependencies=[], 692 gravity_wells={"security": 0.85}, 693 last_updated=datetime.now() 694 )) 695 696 # Thread 3: A4 (Ergodic) focused - different axiom 697 detector.update_thread(ThreadSnapshot( 698 thread_id="deployment-strategy", 699 patterns={"ruin", "catastrophe", "reversible", "rollback", "cheap"}, 700 conclusions=["Deploy decisions must consider ruin"], 701 dependencies=["architecture-review"], 702 gravity_wells={"deploy": 0.8}, 703 last_updated=datetime.now() 704 )) 705 706 print("Thread axiom resonance:") 707 for tid, snap in detector._thread_snapshots.items(): 708 dominant = snap.dominant_axiom 709 if dominant: 710 score = snap.axiom_resonance.get(dominant, 0) 711 name = AXIOM_FIELDS[dominant]["name"] 712 print(f" {tid}: {dominant} ({name}) - {score:.0%}") 713 else: 714 print(f" {tid}: no dominant axiom") 715 716 print() 717 resonances = detector.detect() 718 719 print(f"Detected {len(resonances)} resonances:") 720 for r in resonances: 721 axiom_tag = f" [{r.axiom_type}]" if r.axiom_type else "" 722 print(f" - {r.type.value}{axiom_tag}: {r.pattern}") 723 print(f" strength: {r.strength:.2f}, urgency: {r.urgency:.2f}") 724 if r.axiom_type: 725 print(f" typed as {r.axiom_type} ({r.axiom_name}) @ {r.axiom_confidence:.0%}") 726 if r.should_bubble: 727 print(" ^ SHOULD BUBBLE TO MAIN THREAD") 728 print() 729 730 print("=== Test Complete ===")