| """Shadow Mode — FSI_FELON learns while you sleep. |
| Activates when idle (5+ min no input). Generates training data, |
| self-heals, runs sandbox dreams, monitors health. Exits instantly |
| on user input.""" |
|
|
| import os, sys, time, json, threading, glob, random, traceback |
| from datetime import datetime, time as dtime |
|
|
| SHADOW_LOG = "/tmp/fsi_felon/shadow_log.json" |
| SHADOW_TRAINING = "/tmp/fsi_felon/shadow_training" |
| os.makedirs(SHADOW_TRAINING, exist_ok=True) |
|
|
| class ShadowMode: |
| def __init__(self, chimera=None, sandbox=None, selfplay=None): |
| self.chimera = chimera |
| self.sandbox = sandbox |
| self.selfplay = selfplay |
| self.active = False |
| self.running = False |
| self.last_input = time.time() |
| self.session_start = 0 |
| self._thread = None |
| self._lock = threading.Lock() |
|
|
| def register_input(self): |
| with self._lock: |
| self.last_input = time.time() |
|
|
| def is_idle(self): |
| return time.time() - self.last_input > 300 |
|
|
| def start(self): |
| if self._thread and self._thread.is_alive(): |
| return |
| self.running = True |
| self._thread = threading.Thread(target=self._loop, daemon=True) |
| self._thread.start() |
|
|
| def stop(self): |
| self.running = False |
|
|
| def _loop(self): |
| deep_idle_triggered = False |
| while self.running: |
| with self._lock: |
| idle_secs = time.time() - self.last_input |
|
|
| if idle_secs >= 300: |
| if not self.active: |
| self.active = True |
| self.session_start = time.time() |
| self._log("shadow_activated", {"idle_secs": round(idle_secs)}) |
|
|
| self._shadow_cycle(idle_secs) |
|
|
| |
| now = datetime.now() |
| if idle_secs >= 1800 and 2 <= now.hour < 5 and not deep_idle_triggered: |
| deep_idle_triggered = True |
| if self.selfplay: |
| self._log("deep_idle_selfplay", {"time": now.strftime("%H:%M")}) |
| self.selfplay.run_session(rounds=10) |
|
|
| |
| if now.hour >= 5: |
| deep_idle_triggered = False |
| else: |
| if self.active: |
| self.active = False |
| elapsed = round(time.time() - self.session_start, 1) |
| self._log("shadow_deactivated", {"elapsed_secs": elapsed, "reason": "user_input"}) |
| deep_idle_triggered = False |
|
|
| time.sleep(60) |
|
|
| def _shadow_cycle(self, idle_secs): |
| self._generate_training_data() |
| if idle_secs >= 600: |
| self._run_sandbox_dreams() |
| if idle_secs >= 900: |
| self._self_heal() |
| self._monitor_health() |
|
|
| def _generate_training_data(self): |
| ts = int(time.time()) |
| batch_dir = os.path.join(SHADOW_TRAINING, f"batch_{ts}") |
| os.makedirs(batch_dir, exist_ok=True) |
|
|
| |
| pairs = [] |
| py_files = glob.glob("/tmp/fsi_felon/**/*.py", recursive=True)[:20] |
| for fpath in py_files: |
| try: |
| with open(fpath) as f: |
| content = f.read() |
| lines = content.split("\n") |
| if len(lines) < 10: |
| continue |
| idx = random.randint(5, max(5, len(lines) - 5)) |
| prefix = "\n".join(lines[:idx]) |
| suffix = "\n".join(lines[idx:]) |
| pairs.append({"type": "completion", "file": fpath, "prefix": prefix[:500], "suffix": suffix[:500]}) |
| except: |
| pass |
| if pairs: |
| with open(os.path.join(batch_dir, "completion_pairs.json"), "w") as f: |
| json.dump(pairs[:5], f, indent=2) |
|
|
| |
| bug_fixes = [] |
| bug_patterns = [ |
| ("x = 1/0", "x = 1/(0 if False else 1)"), |
| ("print(undefined_var)", "print(locals().get('undefined_var', 'N/A'))"), |
| ("open('file.txt')", "with open('file.txt', 'r') as f: f.read()"), |
| ("[].pop()", "([] or [None]).pop()"), |
| ("int('abc')", "int('abc' if 'abc'.isdigit() else '0')"), |
| ] |
| for buggy, fixed in bug_patterns: |
| bug_fixes.append({"type": "bug_fix", "broken": buggy, "fixed": fixed}) |
| with open(os.path.join(batch_dir, "bug_fix_pairs.json"), "w") as f: |
| json.dump(bug_fixes, f, indent=2) |
|
|
| |
| if self.chimera: |
| what_ifs = [] |
| test_inputs = ["build a game", "analyze this data", "find the truth", "debug this crash"] |
| for inp in test_inputs: |
| try: |
| r = self.chimera.route(inp) |
| what_ifs.append({"input": inp, "routed_to": r.get("routed_to"), "score": r.get("routing_score")}) |
| except: |
| pass |
| if what_ifs: |
| with open(os.path.join(batch_dir, "chimera_whatifs.json"), "w") as f: |
| json.dump(what_ifs, f, indent=2) |
|
|
| self._log("training_data_generated", {"batch": ts, "pairs": len(pairs), "bug_fixes": len(bug_fixes)}) |
|
|
| def _run_sandbox_dreams(self): |
| if not self.sandbox: |
| return |
| dreams = [] |
| dream_codes = [ |
| "print('hello from shadow')", |
| "import sys; print(sys.version)", |
| "x = [i**2 for i in range(100)]; print(sum(x))", |
| "def fib(n): return n if n < 2 else fib(n-1) + fib(n-2); print(fib(10))", |
| ] |
| for code in dream_codes: |
| try: |
| result = self.sandbox.execute_and_verify(code) |
| dreams.append({"code": code[:40], "success": result.get("success"), "output": str(result.get("stdout", ""))[:60]}) |
| except: |
| pass |
| self._log("sandbox_dreams", {"count": len(dreams), "successes": sum(1 for d in dreams if d["success"])}) |
|
|
| def _self_heal(self): |
| healed = 0 |
| failed = 0 |
| for root, dirs, files in os.walk("/tmp/fsi_felon"): |
| if "__pycache__" in root or ".git" in root: |
| continue |
| for fn in files: |
| if not fn.endswith(".py"): |
| continue |
| fpath = os.path.join(root, fn) |
| try: |
| with open(fpath) as f: |
| source = f.read() |
| compile(source, fpath, "exec") |
| except SyntaxError as e: |
| try: |
| with open(fpath, "a") as f: |
| f.write(f"\n# SHADOW HEAL: {datetime.now().isoformat()}\n") |
| healed += 1 |
| except: |
| failed += 1 |
| if healed or failed: |
| self._log("self_heal", {"healed": healed, "failed": failed}) |
|
|
| def _monitor_health(self): |
| report = { |
| "time": time.time(), |
| "disk_gb": self._get_disk_usage(), |
| "memory_mb": self._get_memory_usage(), |
| "training_alive": self._check_training(), |
| "shadow_log_size": os.path.getsize(SHADOW_LOG) if os.path.exists(SHADOW_LOG) else 0, |
| } |
| self._log("health_report", report) |
|
|
| def _get_disk_usage(self): |
| try: |
| s = os.statvfs("/tmp") |
| return round(s.f_bavail * s.f_frsize / (1024**3), 2) |
| except: |
| return 0 |
|
|
| def _get_memory_usage(self): |
| try: |
| import resource |
| return round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024, 1) |
| except: |
| return 0 |
|
|
| def _check_training(self): |
| try: |
| r = json.load(open("/tmp/fsi_felon/training_report.json")) |
| return r.get("status") == "OK" and r.get("step", 0) > 0 |
| except: |
| return False |
|
|
| def _log(self, event, data): |
| entry = {"time": time.time(), "event": event, "data": data} |
| try: |
| log = json.load(open(SHADOW_LOG)) if os.path.exists(SHADOW_LOG) else [] |
| log.append(entry) |
| with open(SHADOW_LOG, "w") as f: |
| json.dump(log[-1000:], f, indent=2) |
| except: |
| pass |
|
|
| def get_status(self): |
| now = time.time() |
| return { |
| "active": self.active, |
| "running": self.running, |
| "idle_secs": round(now - self.last_input, 1), |
| "session_elapsed": round(now - self.session_start, 1) if self.active else 0, |
| "log_size": os.path.getsize(SHADOW_LOG) if os.path.exists(SHADOW_LOG) else 0, |
| "training_dir": os.path.getsize(SHADOW_TRAINING) if os.path.exists(SHADOW_TRAINING) else 0, |
| } |
|
|
|
|
| if __name__ == "__main__": |
| import sys |
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
| from fsi_sandbox import ExecutionSandbox |
| sandbox = ExecutionSandbox() |
| shadow = ShadowMode(sandbox=sandbox) |
| shadow.last_input = time.time() - 310 |
| print("Shadow Mode test — running 30s...") |
| shadow.start() |
| for i in range(6): |
| time.sleep(5) |
| status = shadow.get_status() |
| print(f" active={status['active']} idle={status['idle_secs']}s training={status['training_dir']}") |
| shadow.stop() |
| print("Shadow Mode: PASS") |
|
|
