spin_up.py
1 #!/usr/bin/env python3 2 """ 3 Spin Up Protocol - Session Initialization 4 5 This is the incantation for starting a new Sovereign OS session. 6 Run this at the start of every Claude Code session. 7 8 Usage: 9 python3 scripts/spin_up.py # Full spin up 10 python3 scripts/spin_up.py --quick # Quick check only 11 python3 scripts/spin_up.py --json # JSON output for programmatic use 12 13 What it does: 14 1. Runs hygiene checks (phoenix state, insight backlog) 15 2. Shows session context (gravity wells, active threads) 16 3. Initializes close down report tracking 17 4. Outputs startup message for agent context 18 19 This script should be run at the START of every session. 20 The close_down.py script should be run at the END. 21 """ 22 23 import argparse 24 import json 25 import sys 26 from datetime import datetime 27 from pathlib import Path 28 29 # Add parent to path 30 sys.path.insert(0, str(Path(__file__).parent.parent)) 31 32 from scripts.phoenix_hygiene import check_all as check_hygiene, format_output 33 34 # Import protocol enforcement (belt and suspenders) 35 try: 36 from scripts.protocol_enforcement import ( 37 run_preflight, 38 phoenix_extraction_gate, 39 test_before_ship, 40 ) 41 HAS_PROTOCOL_ENFORCEMENT = True 42 except ImportError: 43 HAS_PROTOCOL_ENFORCEMENT = False 44 45 # Import resonance lifecycle (context bootstrap) 46 try: 47 from core.attention.resonance_lifecycle import ( 48 create_lifecycle_manager, 49 ResonanceLifecycleManager, 50 ) 51 HAS_RESONANCE_LIFECYCLE = True 52 except ImportError: 53 HAS_RESONANCE_LIFECYCLE = False 54 55 56 SOVEREIGN_OS_ROOT = Path(__file__).parent.parent 57 SESSIONS_DIR = SOVEREIGN_OS_ROOT / "sessions" 58 59 60 def get_live_compression_state() -> dict: 61 """Extract current state from LIVE-COMPRESSION.md.""" 62 live_file = SESSIONS_DIR / "LIVE-COMPRESSION.md" 63 if not live_file.exists(): 64 return {"error": "LIVE-COMPRESSION.md not found"} 65 66 content = live_file.read_text() 67 state = { 68 "session_id": None, 69 "checkpoint": 0, 70 "free_energy": None, 71 "status": None, 72 "gravity_wells": [], 73 "key_insights": [], 74 } 75 76 # Parse metadata 77 for line in content.split('\n'): 78 if 'updated::' in line: 79 state['last_updated'] = line.split('::')[1].strip() 80 elif 'checkpoint::' in line: 81 try: 82 state['checkpoint'] = int(line.split('::')[1].strip().split()[0]) 83 except: 84 pass 85 elif 'free_energy::' in line: 86 try: 87 state['free_energy'] = line.split('=')[1].strip().split()[0] 88 except: 89 pass 90 elif 'status::' in line: 91 state['status'] = line.split('::')[1].strip() 92 elif line.startswith('# Live Compression - '): 93 state['session_id'] = line.replace('# Live Compression - ', '').strip() 94 95 # Extract gravity wells 96 in_wells = False 97 for line in content.split('\n'): 98 if '## Gravity Wells' in line: 99 in_wells = True 100 continue 101 if in_wells: 102 if line.startswith('## '): 103 break 104 if '[[' in line and ']]' in line: 105 # Extract well name 106 well = line.split('[[')[1].split(']]')[0] 107 state['gravity_wells'].append(well) 108 109 return state 110 111 112 def get_insight_backlog_summary() -> dict: 113 """Get summary of insight backlog.""" 114 backlog_file = SESSIONS_DIR / "INSIGHT-BACKLOG.md" 115 if not backlog_file.exists(): 116 return {"count": 0, "high_priority": 0, "items": []} 117 118 content = backlog_file.read_text() 119 lines = content.split('\n') 120 121 items = [] 122 high_priority = 0 123 124 for line in lines: 125 if line.startswith('- [') or line.startswith('* ['): 126 items.append(line) 127 if '[HIGH]' in line or '🔴' in line or 'URGENT' in line: 128 high_priority += 1 129 130 return { 131 "count": len(items), 132 "high_priority": high_priority, 133 "items": items[:5] # First 5 items 134 } 135 136 137 def get_recent_synthesis() -> dict: 138 """Get most recent daily synthesis summary.""" 139 synthesis_file = SESSIONS_DIR / "DAILY-SYNTHESIS.md" 140 if not synthesis_file.exists(): 141 return {"available": False} 142 143 content = synthesis_file.read_text() 144 145 # Get first few lines as summary 146 lines = content.split('\n')[:20] 147 148 return { 149 "available": True, 150 "preview": '\n'.join(lines) 151 } 152 153 154 def initialize_session_tracking(session_name: str = None) -> dict: 155 """Initialize tracking for this session.""" 156 if session_name is None: 157 session_name = datetime.now().strftime("%Y-%m-%d-%H%M") 158 159 # Create close down reports directory if needed 160 reports_dir = SESSIONS_DIR / "close-down-reports" 161 reports_dir.mkdir(exist_ok=True) 162 163 return { 164 "session_id": session_name, 165 "started_at": datetime.now().isoformat(), 166 "tracking_initialized": True, 167 } 168 169 170 def format_startup_message( 171 hygiene_results: list, 172 live_state: dict, 173 backlog: dict, 174 session_tracking: dict 175 ) -> str: 176 """Format the full startup message.""" 177 lines = [] 178 179 # Header 180 lines.append("━" * 74) 181 lines.append("SOVEREIGN OS - SESSION SPIN UP") 182 lines.append("━" * 74) 183 lines.append("") 184 185 # Hygiene Status 186 if not hygiene_results: 187 lines.append("✓ HYGIENE: All checks pass") 188 else: 189 lines.append("⚠️ HYGIENE ISSUES DETECTED:") 190 for name, status, msg in hygiene_results: 191 icon = "🚨" if status == "critical" else "⚠️" 192 lines.append(f" {icon} [{name}] {msg}") 193 lines.append("") 194 195 # Live State 196 lines.append("─" * 74) 197 lines.append("PREVIOUS SESSION STATE") 198 lines.append("─" * 74) 199 if live_state.get('error'): 200 lines.append(f" {live_state['error']}") 201 else: 202 lines.append(f" Session: {live_state.get('session_id', 'unknown')}") 203 lines.append(f" Checkpoint: {live_state.get('checkpoint', 0)}") 204 lines.append(f" Free Energy: {live_state.get('free_energy', 'unknown')}") 205 lines.append(f" Status: {live_state.get('status', 'unknown')}") 206 if live_state.get('gravity_wells'): 207 lines.append(f" Gravity Wells: {', '.join(live_state['gravity_wells'][:5])}") 208 lines.append("") 209 210 # Insight Backlog 211 lines.append("─" * 74) 212 lines.append("INSIGHT BACKLOG") 213 lines.append("─" * 74) 214 if backlog['count'] == 0: 215 lines.append(" (empty)") 216 else: 217 lines.append(f" Total items: {backlog['count']}") 218 if backlog['high_priority'] > 0: 219 lines.append(f" 🔴 HIGH PRIORITY: {backlog['high_priority']}") 220 for item in backlog['items'][:3]: 221 lines.append(f" {item[:70]}") 222 lines.append("") 223 224 # Session Tracking 225 lines.append("─" * 74) 226 lines.append("THIS SESSION") 227 lines.append("─" * 74) 228 lines.append(f" Session ID: {session_tracking['session_id']}") 229 lines.append(f" Started: {session_tracking['started_at']}") 230 lines.append("") 231 232 # Instructions 233 lines.append("─" * 74) 234 lines.append("PROTOCOL REMINDERS") 235 lines.append("─" * 74) 236 lines.append(" • Update LIVE-COMPRESSION.md at every significant checkpoint") 237 lines.append(" • Run close_down.py at session end for full accounting") 238 lines.append(" • Escalate when confidence < 0.4 or scope is wide") 239 lines.append(" • Torah (compressed reports) + Talmud (LIVE-COMPRESSION.md)") 240 lines.append("") 241 242 # File Locations 243 lines.append("─" * 74) 244 lines.append("KEY FILES") 245 lines.append("─" * 74) 246 lines.append(f" Phoenix: {SESSIONS_DIR / 'LIVE-COMPRESSION.md'}") 247 lines.append(f" Backlog: {SESSIONS_DIR / 'INSIGHT-BACKLOG.md'}") 248 lines.append(f" Synthesis: {SESSIONS_DIR / 'DAILY-SYNTHESIS.md'}") 249 lines.append("") 250 251 # Resonance Context (if available) 252 if HAS_RESONANCE_LIFECYCLE: 253 lines.append("─" * 74) 254 lines.append("RESONANCE CONTEXT (Bootstrapped)") 255 lines.append("─" * 74) 256 try: 257 manager = create_lifecycle_manager(str(SESSIONS_DIR), auto_start=False) 258 snapshot = manager.get_latest_snapshot() 259 if snapshot: 260 lines.append(f" Last update: {snapshot.timestamp.strftime('%H:%M:%S')}") 261 lines.append(f" Checkpoint: {snapshot.checkpoint}") 262 if snapshot.dominant_axiom: 263 lines.append(f" Dominant Axiom: {snapshot.dominant_axiom} ({snapshot.dominant_confidence:.0%})") 264 if snapshot.axiom_scores: 265 scores = [f"{k}:{v:.0%}" for k, v in sorted(snapshot.axiom_scores.items(), key=lambda x: -x[1])[:3]] 266 lines.append(f" Top Axioms: {', '.join(scores)}") 267 # Trigger post-compaction bootstrap if this is a new session 268 manager.post_compaction_bootstrap() 269 else: 270 lines.append(" (No previous resonance state)") 271 except Exception as e: 272 lines.append(f" (Resonance bootstrap error: {e})") 273 lines.append("") 274 275 lines.append("━" * 74) 276 lines.append("READY FOR INPUT") 277 lines.append("━" * 74) 278 279 return '\n'.join(lines) 280 281 282 def spin_up(quick: bool = False, json_output: bool = False, session_name: str = None) -> int: 283 """ 284 Main spin up function. 285 286 Returns: 287 0 = clean start 288 1 = warnings present 289 2 = critical issues 290 """ 291 # 1. Run hygiene checks 292 hygiene_results = check_hygiene() 293 294 # 2. Get current state 295 live_state = get_live_compression_state() 296 backlog = get_insight_backlog_summary() 297 298 # 3. Initialize session tracking 299 session_tracking = initialize_session_tracking(session_name) 300 301 # 4. Determine exit code 302 has_critical = any(s == 'critical' for _, s, _ in hygiene_results) 303 has_warnings = any(s == 'warning' for _, s, _ in hygiene_results) 304 305 if json_output: 306 output = { 307 "hygiene": [ 308 {"check": n, "status": s, "message": m} 309 for n, s, m in hygiene_results 310 ], 311 "live_state": live_state, 312 "backlog": backlog, 313 "session": session_tracking, 314 "exit_code": 2 if has_critical else (1 if has_warnings else 0), 315 } 316 print(json.dumps(output, indent=2, default=str)) 317 else: 318 if quick: 319 # Quick mode - just hygiene 320 output = format_output(hygiene_results, 'text') 321 if output: 322 print(output) 323 else: 324 print("✓ All hygiene checks pass - ready to proceed") 325 else: 326 # Full startup message 327 print(format_startup_message( 328 hygiene_results, 329 live_state, 330 backlog, 331 session_tracking 332 )) 333 334 if has_critical: 335 return 2 336 elif has_warnings: 337 return 1 338 return 0 339 340 341 def main(): 342 parser = argparse.ArgumentParser( 343 description="Spin Up Protocol - Session Initialization", 344 formatter_class=argparse.RawDescriptionHelpFormatter, 345 epilog=""" 346 Examples: 347 %(prog)s Full spin up with all context 348 %(prog)s --quick Quick hygiene check only 349 %(prog)s --json JSON output for programmatic use 350 %(prog)s --name my-session Name this session explicitly 351 """ 352 ) 353 354 parser.add_argument('--quick', '-q', action='store_true', 355 help='Quick mode - hygiene check only') 356 parser.add_argument('--json', '-j', action='store_true', 357 help='Output in JSON format') 358 parser.add_argument('--name', '-n', type=str, 359 help='Explicit session name') 360 361 args = parser.parse_args() 362 363 exit_code = spin_up( 364 quick=args.quick, 365 json_output=args.json, 366 session_name=args.name 367 ) 368 sys.exit(exit_code) 369 370 371 if __name__ == "__main__": 372 main()