end_of_day.py
1 #!/usr/bin/env python3 2 """ 3 End-of-Day Protocol - CONSOLIDATED Sovereign OS Shutdown Procedure 4 5 Merges functionality from: 6 - end_of_day_synthesis.py (graph, code health, hunting parties) 7 - session_report.py (value attribution, wisdom capture, economics) 8 - F-value tracking, state saving for next-session continuity 9 10 This is THE ONE end-of-day script. Run before shutting down. 11 12 Usage: 13 python3 scripts/end_of_day.py # Full report 14 python3 scripts/end_of_day.py --quick # Summary only 15 python3 scripts/end_of_day.py --json # Machine-readable 16 """ 17 18 import subprocess 19 import sys 20 import json 21 from pathlib import Path 22 from datetime import datetime 23 from typing import Dict, List, Any, Optional 24 25 REPO_ROOT = Path(__file__).parent.parent 26 SESSIONS_DIR = REPO_ROOT / "sessions" 27 SOVEREIGN_HOME = Path.home() / ".sovereign" 28 29 # ═══════════════════════════════════════════════════════════════════════════════ 30 # DATA COLLECTION 31 # ═══════════════════════════════════════════════════════════════════════════════ 32 33 def run_script(script_path: str, args: List[str] = None, timeout: int = 60) -> str: 34 """Run a script and return output.""" 35 cmd = ["python3", str(REPO_ROOT / script_path)] 36 if args: 37 cmd.extend(args) 38 try: 39 result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) 40 return result.stdout + result.stderr 41 except Exception as e: 42 return f"Error: {e}" 43 44 45 def get_git_commits_today() -> List[Dict[str, str]]: 46 """Get all git commits from today.""" 47 today = datetime.now().strftime("%Y-%m-%d") 48 try: 49 result = subprocess.run( 50 ["git", "log", "--oneline", f"--since={today} 00:00", "--until=tomorrow"], 51 capture_output=True, text=True, cwd=REPO_ROOT 52 ) 53 commits = [] 54 for line in result.stdout.strip().split('\n'): 55 if line: 56 parts = line.split(' ', 1) 57 if len(parts) == 2: 58 commits.append({"hash": parts[0], "message": parts[1]}) 59 return commits 60 except Exception: 61 return [] 62 63 64 def get_debrief_count_today() -> int: 65 """Count today's debrief files.""" 66 today = datetime.now().strftime("%Y-%m-%d") 67 debriefs_dir = SESSIONS_DIR / "debriefs" 68 if not debriefs_dir.exists(): 69 return 0 70 return len(list(debriefs_dir.glob(f"{today}*.md"))) 71 72 73 def get_fo_state() -> Dict[str, Any]: 74 """Read FO-STATE.json for economics and insights.""" 75 fo_state_path = SESSIONS_DIR / "FO-STATE.json" 76 if not fo_state_path.exists(): 77 return {} 78 try: 79 return json.loads(fo_state_path.read_text()) 80 except Exception: 81 return {} 82 83 84 def get_graph_stats() -> Dict: 85 """Get current graph statistics.""" 86 graph_data = SOVEREIGN_HOME / "graph-data.json" 87 if graph_data.exists(): 88 try: 89 data = json.load(open(graph_data)) 90 nodes = data.get("nodes", []) 91 edges = data.get("edges", []) 92 93 # Count by type 94 type_counts = {} 95 for node in nodes: 96 ntype = node.get("type", "unknown") 97 type_counts[ntype] = type_counts.get(ntype, 0) + 1 98 99 # Count orphans 100 connected = set() 101 for edge in edges: 102 connected.add(edge.get("source")) 103 connected.add(edge.get("target")) 104 105 orphan_ids = set(n.get("id") for n in nodes) - connected 106 orphan_rate = len(orphan_ids) / len(nodes) if nodes else 0 107 108 # Count sources 109 sources = set() 110 for node in nodes: 111 src = node.get("source", "") 112 if src: 113 sources.add(src.split(":")[0] if ":" in src else src) 114 115 return { 116 "nodes": len(nodes), 117 "edges": len(edges), 118 "orphans": len(orphan_ids), 119 "orphan_rate": orphan_rate, 120 "sources": len(sources), 121 "type_counts": type_counts 122 } 123 except: 124 pass 125 return {"nodes": 0, "edges": 0, "orphans": 0, "orphan_rate": 0, "sources": 0, "type_counts": {}} 126 127 128 def get_code_health() -> Dict: 129 """Get code health metrics from daily_code_synthesis.""" 130 try: 131 output = run_script("scripts/daily_code_synthesis.py", ["--json"], timeout=30) 132 # Try to parse JSON from output 133 for line in output.split('\n'): 134 if line.strip().startswith('{'): 135 return json.loads(line) 136 except: 137 pass 138 139 # Fallback: run basic checks 140 health = {"duplicates": 0, "dead_imports": 0, "similar_clusters": 0} 141 try: 142 # Count Python files 143 py_files = list(REPO_ROOT.glob("**/*.py")) 144 py_files = [f for f in py_files if "__pycache__" not in str(f) and ".venv" not in str(f)] 145 health["python_files"] = len(py_files) 146 except: 147 pass 148 return health 149 150 151 def calculate_f_deltas(insights: List[Dict]) -> Dict[str, float]: 152 """Calculate total F-value improvements from insights.""" 153 total_delta = 0.0 154 start_f = 0.15 155 end_f = 0.15 156 157 for insight in insights: 158 content = insight.get("content", "") 159 ctx = insight.get("context", "") 160 161 # Extract delta 162 if "ΔF = -" in content: 163 try: 164 delta = float(content.split("ΔF = -")[1].split()[0]) 165 total_delta += delta 166 except (ValueError, IndexError): 167 pass 168 169 # Extract F values 170 for text in [content, ctx]: 171 if "F:" in text and "→" in text: 172 try: 173 parts = text.split("F:")[1].split("→") 174 if len(parts) >= 2: 175 start_f = float(parts[0].strip().split()[0]) 176 end_f = float(parts[1].strip().split()[0]) 177 except (ValueError, IndexError): 178 pass 179 180 return { 181 "total_delta": round(total_delta, 2), 182 "start_f": start_f, 183 "end_f": end_f, 184 "net_improvement": round(start_f - end_f, 2) 185 } 186 187 188 def get_session_state() -> Dict[str, Any]: 189 """Load session report state (insights, resonances, etc).""" 190 state_file = SOVEREIGN_HOME / "session-report-state.json" 191 if state_file.exists(): 192 try: 193 return json.loads(state_file.read_text()) 194 except: 195 pass 196 return {} 197 198 199 # ═══════════════════════════════════════════════════════════════════════════════ 200 # REPORT GENERATION 201 # ═══════════════════════════════════════════════════════════════════════════════ 202 203 def generate_report() -> Dict[str, Any]: 204 """Generate comprehensive end-of-day report data.""" 205 fo_state = get_fo_state() 206 commits = get_git_commits_today() 207 debrief_count = get_debrief_count_today() 208 graph = get_graph_stats() 209 code_health = get_code_health() 210 session_state = get_session_state() 211 212 # Economics 213 costs = fo_state.get("costs", {}) 214 savings = costs.get("savings", {}) 215 efficiency = costs.get("efficiency", {}) 216 sats = costs.get("sats", {}) 217 218 # F-value tracking 219 insights = fo_state.get("insights", []) 220 f_deltas = calculate_f_deltas(insights) 221 222 # Axiom activity 223 axiom_activity = fo_state.get("axiom_activity", {}) 224 225 return { 226 "date": datetime.now().strftime("%Y-%m-%d"), 227 "generated": datetime.now().isoformat(), 228 229 # Graph State 230 "graph": { 231 "nodes": graph["nodes"], 232 "edges": graph["edges"], 233 "orphans": graph["orphans"], 234 "orphan_rate": graph["orphan_rate"], 235 "sources": graph["sources"], 236 }, 237 238 # Alignment (F-value) 239 "alignment": { 240 "f_start": f_deltas["start_f"], 241 "f_end": f_deltas["end_f"], 242 "delta_f": f_deltas["net_improvement"], 243 "insights_count": len(insights), 244 }, 245 246 # Economics 247 "economics": { 248 "subscription_sats": sats.get("subscription_sats", 200000), 249 "subscription_usd": costs.get("usd", {}).get("subscription", 200.00), 250 "api_equivalent_sats": sats.get("api_sats", 0), 251 "api_equivalent_usd": costs.get("usd", {}).get("api", 0), 252 "savings_vs_api_sats": savings.get("vs_api_sats", 0), 253 "savings_vs_api_usd": savings.get("vs_api_usd", 0), 254 "api_efficiency": efficiency.get("api_efficiency", 1.0), 255 "infra_efficiency": efficiency.get("infra_efficiency", 1.0), 256 }, 257 258 # Wisdom captured 259 "wisdom": { 260 "insights": len(session_state.get("insights", [])), 261 "resonances": len(session_state.get("resonances", [])), 262 "principle_edges": len(session_state.get("principle_edges", [])), 263 "done_items": len(session_state.get("done_items", [])), 264 }, 265 266 # Work completed 267 "work": { 268 "commits": len(commits), 269 "commit_list": commits[:15], 270 "debriefs": debrief_count, 271 }, 272 273 # Code health 274 "code_health": code_health, 275 276 # Axiom activity 277 "axioms": { 278 "total": sum(axiom_activity.values()), 279 "breakdown": axiom_activity, 280 }, 281 } 282 283 284 def print_report(report: Dict[str, Any], quick: bool = False): 285 """Print human-readable end-of-day report.""" 286 date = report["date"] 287 288 print() 289 print(" ╔═══════════════════════════════════════════════════════════════╗") 290 print(f" ║ {date} END OF DAY ║") 291 print(" ╠═══════════════════════════════════════════════════════════════╣") 292 print(" ║ ║") 293 294 # Graph State 295 g = report["graph"] 296 print(" ║ GRAPH STATE ║") 297 print(f" ║ Nodes: {g['nodes']:,}".ljust(63) + "║") 298 print(f" ║ Edges: {g['edges']:,}".ljust(63) + "║") 299 print(f" ║ Data sources: {g['sources']}".ljust(63) + "║") 300 print(" ║ ║") 301 302 # Wisdom 303 w = report["wisdom"] 304 ax = report["axioms"] 305 print(" ║ WISDOM CAPTURED ║") 306 print(f" ║ Insights: {report['alignment']['insights_count']} today".ljust(63) + "║") 307 print(f" ║ Debriefs: {report['work']['debriefs']} session reports".ljust(63) + "║") 308 axiom_str = " ".join([f"{k}:{v}" for k, v in sorted(ax['breakdown'].items())]) 309 print(f" ║ Axiom activity: {axiom_str}".ljust(63) + "║") 310 print(" ║ ║") 311 312 # Code Health 313 ch = report.get("code_health", {}) 314 if ch.get("duplicates", 0) > 0 or ch.get("dead_imports", 0) > 0: 315 print(" ║ CODE HEALTH (needs attention tomorrow) ║") 316 if ch.get("duplicates"): 317 print(f" ║ Duplicate functions: {ch['duplicates']}".ljust(63) + "║") 318 if ch.get("dead_imports"): 319 print(f" ║ Dead imports: {ch['dead_imports']}".ljust(63) + "║") 320 if ch.get("similar_clusters"): 321 print(f" ║ Similar file clusters: {ch['similar_clusters']}".ljust(63) + "║") 322 print(" ║ ║") 323 324 if not quick: 325 # Economics 326 econ = report["economics"] 327 print(" ║ ECONOMICS ║") 328 print(f" ║ Subscription: {econ['subscription_sats']:,} sats (${econ['subscription_usd']:.0f})".ljust(63) + "║") 329 print(f" ║ Savings vs API: {econ['savings_vs_api_sats']:,} sats (${econ['savings_vs_api_usd']:.0f})".ljust(63) + "║") 330 print(f" ║ Efficiency: {econ['api_efficiency']:.1f}x API".ljust(63) + "║") 331 print(" ║ ║") 332 333 # F-value 334 al = report["alignment"] 335 print(" ║ ALIGNMENT ║") 336 print(f" ║ F: {al['f_start']:.2f} → {al['f_end']:.2f} | ΔF = -{al['delta_f']:.2f}".ljust(63) + "║") 337 print(" ║ ║") 338 339 # Commits 340 print(" ║ COMMITS TODAY ║") 341 print(f" ║ {report['work']['commits']} commits".ljust(63) + "║") 342 for commit in report["work"]["commit_list"][:5]: 343 msg = commit["message"][:45] 344 print(f" ║ {commit['hash'][:7]} {msg}".ljust(63) + "║") 345 print(" ║ ║") 346 347 # Reports saved 348 print(" ║ REPORTS SAVED ║") 349 debrief_dir = SESSIONS_DIR / "debriefs" 350 today = datetime.now().strftime("%Y-%m-%d") 351 debriefs = sorted(debrief_dir.glob(f"{today}*.md")) 352 if debriefs: 353 latest = debriefs[-1] 354 print(f" ║ {latest.relative_to(REPO_ROOT)}".ljust(63) + "║") 355 356 synthesis_dir = SESSIONS_DIR / "synthesis" 357 if synthesis_dir.exists(): 358 syntheses = sorted(synthesis_dir.glob(f"*{today}*.md")) 359 if syntheses: 360 latest = syntheses[-1] 361 print(f" ║ {latest.relative_to(REPO_ROOT)}".ljust(63) + "║") 362 363 print(" ║ ║") 364 print(" ╚═══════════════════════════════════════════════════════════════╝") 365 print() 366 print(" Good night.") 367 print() 368 369 370 def save_state(report: Dict[str, Any]) -> Path: 371 """Save state for next-session continuity.""" 372 state_file = SESSIONS_DIR / f"state-{report['date']}.json" 373 state_file.write_text(json.dumps(report, indent=2)) 374 return state_file 375 376 377 def save_synthesis_report(report: Dict[str, Any]) -> Path: 378 """Save code synthesis report to sessions/synthesis/.""" 379 synthesis_dir = SESSIONS_DIR / "synthesis" 380 synthesis_dir.mkdir(exist_ok=True) 381 382 lines = [ 383 f"# Code Synthesis - {report['date']}", 384 "", 385 f"Generated: {report['generated']}", 386 "", 387 "## Graph State", 388 f"- Nodes: {report['graph']['nodes']:,}", 389 f"- Edges: {report['graph']['edges']:,}", 390 f"- Orphan rate: {report['graph']['orphan_rate']:.1%}", 391 f"- Data sources: {report['graph']['sources']}", 392 "", 393 "## Code Health", 394 ] 395 396 ch = report.get("code_health", {}) 397 if ch: 398 for key, value in ch.items(): 399 lines.append(f"- {key}: {value}") 400 else: 401 lines.append("- No issues detected") 402 403 lines.extend([ 404 "", 405 "## Axiom Activity", 406 f"Total: {report['axioms']['total']}", 407 "", 408 ]) 409 for ax, count in sorted(report['axioms']['breakdown'].items()): 410 bar = "█" * min(count // 10, 20) 411 lines.append(f"- {ax}: {count} {bar}") 412 413 filepath = synthesis_dir / f"code-synthesis-{report['date']}.md" 414 filepath.write_text('\n'.join(lines)) 415 return filepath 416 417 418 def main(): 419 quick = "--quick" in sys.argv 420 json_output = "--json" in sys.argv 421 422 report = generate_report() 423 424 if json_output: 425 print(json.dumps(report, indent=2)) 426 return 427 428 # Save outputs 429 save_state(report) 430 save_synthesis_report(report) 431 432 # Run session_report.py to generate Obsidian debrief 433 run_script("scripts/session_report.py", timeout=30) 434 435 # Print consolidated report 436 print_report(report, quick=quick) 437 438 439 if __name__ == "__main__": 440 main()