insight_escalator.py
1 #!/usr/bin/env python3 2 """ 3 Insight Escalator - Detect high-resonance statements and escalate 4 5 This script bridges live conversation to the graph: 6 1. Takes text input (from user statements, conversation) 7 2. Calculates resonance with existing concepts 8 3. If resonance is high, creates an alert for First Officer 9 4. Updates gravity topology 10 11 This implements the user's request: 12 "I don't think you need to manage that live. I think that our system should 13 identify these concepts that have higher resonance and escalate them automatically." 14 15 Usage: 16 echo "philosophy becomes code" | python scripts/insight_escalator.py 17 python scripts/insight_escalator.py --text "build liberally, rewrites are cheap" 18 python scripts/insight_escalator.py --file conversation.txt 19 """ 20 21 import os 22 import re 23 import sys 24 import json 25 from pathlib import Path 26 from datetime import datetime 27 from collections import Counter 28 from typing import Dict, List, Tuple, Optional 29 import argparse 30 31 # Import from resonance engine 32 sys.path.insert(0, str(Path(__file__).parent)) 33 34 SCAN_DIRS = ['docs', 'patterns', 'sessions', 'dashboards'] 35 SKIP_DIRS = {'.git', 'node_modules', '__pycache__', 'templates'} 36 37 STOP_WORDS = { 38 'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 39 'of', 'with', 'by', 'from', 'as', 'is', 'was', 'are', 'were', 'been', 40 'be', 'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 41 'could', 'should', 'may', 'might', 'must', 'shall', 'can', 'need', 42 'this', 'that', 'these', 'those', 'it', 'its', 'they', 'them', 43 'their', 'we', 'us', 'our', 'you', 'your', 'he', 'she', 'him', 'her', 44 'not', 'no', 'yes', 'all', 'any', 'each', 'every', 'both', 'few', 45 'more', 'most', 'other', 'some', 'such', 'only', 'same', 'so', 'than', 46 'too', 'very', 'just', 'also', 'now', 'here', 'there', 'when', 'where', 47 'why', 'how', 'what', 'which', 'who', 'whom', 'if', 'then', 'else', 48 'like', 'think', 'things', 'going', 'want', 'know', 'see', 'get', 49 'make', 'way', 'something', 'really', 'actually', 'maybe', 50 } 51 52 HIGH_VALUE_TERMS = { 53 # Axioms 54 'boundary', 'markov', 'blanket', 'integration', 'telos', 'recognition', 55 'life', 'death', 'navigation', 'pole', 'ergodic', 'ruin', 'survival', 56 # Philosophers 57 'deleuze', 'hayek', 'taleb', 'friston', 'dellanna', 'guattari', 58 # Concepts 59 'rhizome', 'deterritorialization', 'spontaneous', 'order', 'emergence', 60 'entropy', 'free-energy', 'active-inference', 'autopoiesis', 61 'feedback', 'loop', 'adaptation', 'signal', 'compression', 62 # Actions 63 'build', 'implement', 'code', 'rewrite', 'test', 'ship', 64 # Meta 65 'philosophy', 'principle', 'axiom', 'pattern', 'resonance', 'gravity', 66 } 67 68 69 def extract_terms(text: str) -> Counter: 70 """Extract weighted terms from text.""" 71 text = text.lower() 72 words = re.findall(r'[a-zA-Z][a-zA-Z-]*[a-zA-Z]|[a-zA-Z]', text) 73 74 terms = Counter() 75 for word in words: 76 if word in STOP_WORDS or len(word) < 3: 77 continue 78 weight = 3 if word in HIGH_VALUE_TERMS else 1 79 terms[word] += weight 80 81 return terms 82 83 84 def scan_concept_graph(base_path: Path) -> Dict[str, Counter]: 85 """Scan graph and build concept term map.""" 86 concepts = {} 87 88 for scan_dir in SCAN_DIRS: 89 dir_path = base_path / scan_dir 90 if not dir_path.exists(): 91 continue 92 93 for filepath in dir_path.rglob('*.md'): 94 if any(skip in filepath.parts for skip in SKIP_DIRS): 95 continue 96 97 try: 98 content = filepath.read_text(encoding='utf-8') 99 except Exception: 100 continue 101 102 name = filepath.stem.lower().replace(' ', '-') 103 concepts[name] = extract_terms(content) 104 105 return concepts 106 107 108 def calculate_resonance(input_terms: Counter, concept_terms: Counter) -> float: 109 """Calculate resonance between input and concept.""" 110 if not input_terms or not concept_terms: 111 return 0.0 112 113 shared = set(input_terms.keys()) & set(concept_terms.keys()) 114 if not shared: 115 return 0.0 116 117 shared_weight = sum( 118 min(input_terms[term], concept_terms[term]) 119 for term in shared 120 ) 121 122 input_total = sum(input_terms.values()) 123 concept_total = sum(concept_terms.values()) 124 125 if input_total == 0 or concept_total == 0: 126 return 0.0 127 128 normalizer = (input_total * concept_total) ** 0.5 129 score = shared_weight / normalizer 130 131 # Boost for high-value term overlap 132 high_value_shared = shared & HIGH_VALUE_TERMS 133 if high_value_shared: 134 score *= (1 + 0.3 * len(high_value_shared)) 135 136 return min(1.0, score) 137 138 139 def find_resonances( 140 text: str, 141 concepts: Dict[str, Counter], 142 threshold: float = 0.15 143 ) -> List[Tuple[str, float]]: 144 """Find concepts that resonate with input text.""" 145 input_terms = extract_terms(text) 146 if not input_terms: 147 return [] 148 149 resonances = [] 150 for name, terms in concepts.items(): 151 score = calculate_resonance(input_terms, terms) 152 if score >= threshold: 153 resonances.append((name, score)) 154 155 resonances.sort(key=lambda x: x[1], reverse=True) 156 return resonances 157 158 159 def create_escalation_alert( 160 text: str, 161 resonances: List[Tuple[str, float]], 162 alerts_dir: Path 163 ) -> Path: 164 """Create an escalation alert for the First Officer.""" 165 timestamp = datetime.now().strftime('%Y%m%d-%H%M%S') 166 filename = f"{timestamp}-insight-escalation.md" 167 filepath = alerts_dir / filename 168 169 top_resonances = resonances[:5] 170 max_score = top_resonances[0][1] if top_resonances else 0 171 172 urgency = min(1.0, max_score * 1.5) # Scale up 173 174 lines = [ 175 f"# Insight Escalation - {datetime.now().strftime('%Y-%m-%d %H:%M')}", 176 "", 177 f"**Type:** HIGH_RESONANCE_INSIGHT", 178 f"**Detected:** {datetime.now().isoformat()}", 179 f"**Max Resonance:** {max_score:.0%}", 180 f"**Urgency:** {urgency:.2f}", 181 "", 182 "---", 183 "", 184 "## Input Statement", 185 "", 186 f"> {text}", 187 "", 188 "## Resonant Concepts", 189 "", 190 ] 191 192 for concept, score in top_resonances: 193 lines.append(f"- [[{concept}]] - {score:.0%}") 194 195 lines.extend([ 196 "", 197 "## Integration Recommendation", 198 "", 199 ]) 200 201 if max_score > 0.5: 202 lines.append("**HIGH PRIORITY** - This insight has strong resonance with core concepts.") 203 lines.append("Consider integrating into the relevant concept pages.") 204 elif max_score > 0.3: 205 lines.append("**MEDIUM PRIORITY** - Notable resonance detected.") 206 lines.append("Review for potential inclusion in related concepts.") 207 else: 208 lines.append("**LOW PRIORITY** - Some resonance detected.") 209 lines.append("May be tangential or exploratory.") 210 211 lines.extend([ 212 "", 213 "---", 214 "", 215 f"*Generated by Insight Escalator | {datetime.now().strftime('%Y-%m-%d %H:%M')}*", 216 ]) 217 218 filepath.write_text('\n'.join(lines), encoding='utf-8') 219 return filepath 220 221 222 def main(): 223 parser = argparse.ArgumentParser( 224 description='Insight Escalator - Detect high-resonance statements' 225 ) 226 parser.add_argument( 227 '--text', '-t', 228 type=str, 229 help='Text to analyze' 230 ) 231 parser.add_argument( 232 '--file', '-f', 233 type=Path, 234 help='File containing text to analyze' 235 ) 236 parser.add_argument( 237 '--threshold', 238 type=float, 239 default=0.15, 240 help='Minimum resonance threshold (default: 0.15)' 241 ) 242 parser.add_argument( 243 '--escalate', 244 action='store_true', 245 help='Create escalation alert if resonance exceeds threshold' 246 ) 247 parser.add_argument( 248 '--path', '-p', 249 type=Path, 250 default=Path(__file__).parent.parent, 251 help='Path to Sovereign_OS repo' 252 ) 253 254 args = parser.parse_args() 255 256 # Get input text 257 if args.text: 258 text = args.text 259 elif args.file: 260 text = args.file.read_text(encoding='utf-8') 261 elif not sys.stdin.isatty(): 262 text = sys.stdin.read() 263 else: 264 print("Error: No input provided. Use --text, --file, or pipe input.", file=sys.stderr) 265 return 1 266 267 text = text.strip() 268 if not text: 269 print("Error: Empty input", file=sys.stderr) 270 return 1 271 272 # Scan concepts 273 concepts = scan_concept_graph(args.path) 274 if not concepts: 275 print("Error: No concepts found in graph", file=sys.stderr) 276 return 1 277 278 # Find resonances 279 resonances = find_resonances(text, concepts, args.threshold) 280 281 print("=" * 60) 282 print("INSIGHT RESONANCE ANALYSIS") 283 print("=" * 60) 284 print() 285 print(f"Input: \"{text[:80]}{'...' if len(text) > 80 else ''}\"") 286 print(f"Concepts scanned: {len(concepts)}") 287 print() 288 289 if not resonances: 290 print("No significant resonances found.") 291 return 0 292 293 print(f"Found {len(resonances)} resonant concepts:") 294 print() 295 296 for concept, score in resonances[:10]: 297 bar = "█" * int(score * 20) 298 print(f" [{score:>5.0%}] {bar:<20} {concept}") 299 300 print() 301 302 # Check if escalation needed 303 max_score = resonances[0][1] 304 if args.escalate and max_score > 0.25: 305 alerts_dir = args.path / 'sessions' / 'RESONANCE-ALERTS' 306 alerts_dir.mkdir(parents=True, exist_ok=True) 307 308 alert_path = create_escalation_alert(text, resonances, alerts_dir) 309 print(f"✓ Escalation alert created: {alert_path}") 310 elif max_score > 0.25: 311 print(f"⚠ High resonance ({max_score:.0%}) - consider using --escalate") 312 313 return 0 314 315 316 if __name__ == '__main__': 317 sys.exit(main())