close_down.py
1 #!/usr/bin/env python3 2 """ 3 Close Down Protocol - Session Finalization 4 5 This is the exit incantation for ending a Sovereign OS session. 6 Run this at the END of every Claude Code session. 7 8 Usage: 9 python3 scripts/close_down.py # Full close down 10 python3 scripts/close_down.py --quick # Quick summary only 11 python3 scripts/close_down.py --name my-session # Name this session 12 13 What it does: 14 1. Reads LIVE-COMPRESSION.md for session state 15 2. Generates close down report (full accounting) 16 3. Saves report to close-down-reports/ 17 4. Shows summary to user 18 19 This script should be run at the END of every session. 20 The spin_up.py script should be run at the START. 21 """ 22 23 import argparse 24 import json 25 import sys 26 from datetime import datetime 27 from pathlib import Path 28 from typing import Dict, Any, List 29 30 # Import common utilities for portability 31 try: 32 from sos_common import ( 33 find_repo_root, 34 get_sessions_dir, 35 get_live_compression_path, 36 get_close_down_reports_dir, 37 load_config, 38 ) 39 REPO_ROOT = find_repo_root() 40 except ImportError: 41 # Fallback for standalone use 42 REPO_ROOT = Path(__file__).parent.parent 43 44 def get_sessions_dir(r=None): 45 return REPO_ROOT / "sessions" 46 47 def get_live_compression_path(r=None): 48 return get_sessions_dir() / "LIVE-COMPRESSION.md" 49 50 def get_close_down_reports_dir(r=None): 51 return get_sessions_dir() / "close-down-reports" 52 53 def load_config(r=None): 54 return {} 55 56 57 def parse_live_compression() -> Dict[str, Any]: 58 """Parse LIVE-COMPRESSION.md to extract session state.""" 59 live_file = get_live_compression_path(REPO_ROOT) 60 61 if not live_file.exists(): 62 return {"error": "LIVE-COMPRESSION.md not found"} 63 64 content = live_file.read_text() 65 state = { 66 "session_id": "unknown", 67 "checkpoints": 0, 68 "free_energy": "unknown", 69 "status": "unknown", 70 "gravity_wells": [], 71 "key_insights": [], 72 "artifacts": [], 73 "decisions": [], 74 "paths_not_taken": [], 75 } 76 77 current_section = None 78 in_list = False 79 80 for line in content.split('\n'): 81 # Extract metadata 82 if 'updated::' in line: 83 state['last_updated'] = line.split('::')[1].strip() 84 elif 'checkpoint::' in line: 85 try: 86 state['checkpoints'] = int(line.split('::')[1].strip().split()[0]) 87 except: 88 pass 89 elif 'free_energy::' in line: 90 try: 91 # Parse "F = 0.05" format 92 if '=' in line: 93 state['free_energy'] = float(line.split('=')[1].strip().split()[0]) 94 except: 95 pass 96 elif 'status::' in line: 97 state['status'] = line.split('::')[1].strip() 98 elif line.startswith('# Live Compression - '): 99 state['session_id'] = line.replace('# Live Compression - ', '').strip() 100 101 # Track sections 102 if line.startswith('## '): 103 current_section = line.replace('## ', '').strip().lower() 104 in_list = False 105 106 # Extract gravity wells 107 if current_section and 'gravity' in current_section: 108 if '[[' in line and ']]' in line: 109 well = line.split('[[')[1].split(']]')[0] 110 state['gravity_wells'].append(well) 111 112 # Extract key insights 113 if current_section and 'insight' in current_section: 114 if line.strip().startswith(('1.', '2.', '3.', '4.', '5.', '-', '*')): 115 insight = line.strip().lstrip('0123456789.-* ') 116 if insight and len(insight) > 5: 117 state['key_insights'].append(insight) 118 119 # Extract artifacts 120 if current_section and 'artifact' in current_section: 121 if '|' in line and 'Artifact' not in line and '---' not in line: 122 parts = [p.strip() for p in line.split('|') if p.strip()] 123 if len(parts) >= 2: 124 state['artifacts'].append({ 125 'name': parts[0], 126 'type': parts[1] if len(parts) > 1 else 'unknown', 127 'path': parts[2] if len(parts) > 2 else '', 128 }) 129 130 # Extract paths not taken 131 if current_section and 'not taken' in current_section: 132 if line.strip().startswith(('-', '*', '1.', '2.', '3.')): 133 path = line.strip().lstrip('-* 0123456789.') 134 if path and len(path) > 3: 135 state['paths_not_taken'].append(path) 136 137 return state 138 139 140 def generate_close_down_report(state: Dict[str, Any], session_name: str = None) -> str: 141 """Generate the close down report markdown.""" 142 session_id = session_name or state.get('session_id', datetime.now().strftime("%Y-%m-%d-%H%M")) 143 now = datetime.now() 144 145 lines = [] 146 147 # Header 148 lines.append(f"# Close Down Report: {session_id}") 149 lines.append("") 150 lines.append(f"*Generated: {now.strftime('%Y-%m-%d %H:%M')}*") 151 lines.append("") 152 153 # Executive Summary 154 lines.append("---") 155 lines.append("") 156 lines.append("## Executive Summary") 157 lines.append("") 158 159 f_value = state.get('free_energy', 'unknown') 160 if isinstance(f_value, (int, float)): 161 f_status = "aligned" if f_value < 0.1 else ("minor deviation" if f_value < 0.3 else "significant") 162 lines.append(f"| Metric | Value |") 163 lines.append(f"|--------|-------|") 164 lines.append(f"| **Free Energy (F)** | {f_value:.3f} ({f_status}) |") 165 else: 166 lines.append(f"| Metric | Value |") 167 lines.append(f"|--------|-------|") 168 lines.append(f"| **Free Energy (F)** | {f_value} |") 169 170 lines.append(f"| **Phoenix Checkpoints** | {state.get('checkpoints', 0)} |") 171 lines.append(f"| **Artifacts Created** | {len(state.get('artifacts', []))} |") 172 lines.append(f"| **Insights Captured** | {len(state.get('key_insights', []))} |") 173 lines.append(f"| **Status** | {state.get('status', 'unknown')} |") 174 lines.append("") 175 176 # Gravity Wells 177 if state.get('gravity_wells'): 178 lines.append("---") 179 lines.append("") 180 lines.append("## Active Gravity Wells") 181 lines.append("") 182 for well in state['gravity_wells'][:10]: 183 lines.append(f"- [[{well}]]") 184 lines.append("") 185 186 # Artifacts 187 if state.get('artifacts'): 188 lines.append("---") 189 lines.append("") 190 lines.append("## Artifacts Created") 191 lines.append("") 192 lines.append("| Artifact | Type | Path |") 193 lines.append("|----------|------|------|") 194 for art in state['artifacts']: 195 lines.append(f"| {art['name']} | {art['type']} | `{art['path']}` |") 196 lines.append("") 197 198 # Key Insights 199 if state.get('key_insights'): 200 lines.append("---") 201 lines.append("") 202 lines.append("## Key Insights") 203 lines.append("") 204 for i, insight in enumerate(state['key_insights'][:10], 1): 205 lines.append(f"{i}. {insight}") 206 lines.append("") 207 208 # Paths Not Taken 209 if state.get('paths_not_taken'): 210 lines.append("---") 211 lines.append("") 212 lines.append("## Paths Not Taken") 213 lines.append("") 214 for path in state['paths_not_taken']: 215 lines.append(f"- {path}") 216 lines.append("") 217 218 # Footer 219 lines.append("---") 220 lines.append("") 221 lines.append(f"*Session: {session_id} | Closed: {now.isoformat()}*") 222 223 return '\n'.join(lines) 224 225 226 def save_report(report: str, session_name: str) -> Path: 227 """Save the report to the close-down-reports directory.""" 228 reports_dir = get_close_down_reports_dir(REPO_ROOT) 229 reports_dir.mkdir(parents=True, exist_ok=True) 230 231 # Create filename 232 date_str = datetime.now().strftime("%Y-%m-%d") 233 filename = f"{date_str}-{session_name.replace('/', '-')}.md" 234 filepath = reports_dir / filename 235 236 filepath.write_text(report) 237 return filepath 238 239 240 def close_down(session_name: str = None, quick: bool = False, json_output: bool = False) -> int: 241 """ 242 Main close down function. 243 244 Returns: 245 0 = Success 246 1 = Warnings 247 2 = Errors 248 """ 249 print() 250 print("━" * 60) 251 print("SOVEREIGN OS - SESSION CLOSE DOWN") 252 print("━" * 60) 253 print() 254 255 # Parse live compression 256 state = parse_live_compression() 257 258 if state.get('error'): 259 print(f"⚠ {state['error']}") 260 return 2 261 262 session_id = session_name or state.get('session_id', datetime.now().strftime("%Y-%m-%d-%H%M")) 263 264 if quick: 265 # Quick summary 266 print(f"Session: {session_id}") 267 print(f"Checkpoints: {state.get('checkpoints', 0)}") 268 print(f"Free Energy: {state.get('free_energy', 'unknown')}") 269 print(f"Artifacts: {len(state.get('artifacts', []))}") 270 print(f"Insights: {len(state.get('key_insights', []))}") 271 return 0 272 273 if json_output: 274 print(json.dumps(state, indent=2, default=str)) 275 return 0 276 277 # Generate full report 278 report = generate_close_down_report(state, session_id) 279 280 # Save report 281 filepath = save_report(report, session_id) 282 283 print(f"Session: {session_id}") 284 print(f"Status: {state.get('status', 'unknown')}") 285 print() 286 287 # Summary stats 288 print("─" * 60) 289 print("SESSION SUMMARY") 290 print("─" * 60) 291 292 f_val = state.get('free_energy', 'unknown') 293 if isinstance(f_val, (int, float)): 294 status_icon = "✓" if f_val < 0.1 else ("⚠" if f_val < 0.3 else "✗") 295 print(f" {status_icon} Free Energy: F = {f_val:.3f}") 296 else: 297 print(f" ? Free Energy: {f_val}") 298 299 print(f" 📍 Checkpoints: {state.get('checkpoints', 0)}") 300 print(f" 📦 Artifacts: {len(state.get('artifacts', []))}") 301 print(f" 💡 Insights: {len(state.get('key_insights', []))}") 302 print() 303 304 # Show gravity wells 305 if state.get('gravity_wells'): 306 print("─" * 60) 307 print("ACTIVE GRAVITY WELLS") 308 print("─" * 60) 309 for well in state['gravity_wells'][:5]: 310 print(f" • {well}") 311 print() 312 313 # Report saved 314 print("─" * 60) 315 print("REPORT SAVED") 316 print("─" * 60) 317 print(f" {filepath}") 318 print() 319 320 print("━" * 60) 321 print("SESSION CLOSED") 322 print("━" * 60) 323 324 return 0 325 326 327 def main(): 328 parser = argparse.ArgumentParser( 329 description="Close Down Protocol - Session Finalization", 330 formatter_class=argparse.RawDescriptionHelpFormatter, 331 epilog=""" 332 Examples: 333 %(prog)s Full close down with report 334 %(prog)s --quick Quick summary only 335 %(prog)s --name my-session Name this session explicitly 336 %(prog)s --json JSON output 337 """ 338 ) 339 340 parser.add_argument('--quick', '-q', action='store_true', 341 help='Quick summary only') 342 parser.add_argument('--json', '-j', action='store_true', 343 help='Output in JSON format') 344 parser.add_argument('--name', '-n', type=str, 345 help='Explicit session name') 346 347 args = parser.parse_args() 348 349 exit_code = close_down( 350 session_name=args.name, 351 quick=args.quick, 352 json_output=args.json 353 ) 354 sys.exit(exit_code) 355 356 357 if __name__ == "__main__": 358 main()