File size: 2,914 Bytes
fc2988c 29cdc9d fc2988c 29cdc9d fc2988c 29cdc9d fc2988c 29cdc9d fc2988c 29cdc9d fc2988c 29cdc9d fc2988c 29cdc9d fc2988c 29cdc9d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | import json, os, time
from src.ide_kernel.kernel import SovereignKernel
from src.ide_kernel.validator import KernelValidator
from src.ide_kernel.ledger import ProjectLedger
from src.brain.resonance import ResonanceEngine
class KernelDaemon:
def __init__(self, workspace_path):
self.root = os.path.abspath(workspace_path)
self.task_file = os.path.join(self.root, "workspace_tasks.json")
self.kernel = SovereignKernel(self.root)
self.ledger = ProjectLedger(self.root)
self.resonance = ResonanceEngine()
def handle_failure(self, task, output):
path = os.path.join(self.root, "failure_report.json")
with open(path, 'w') as f:
json.dump({"intent":"auto_debug","original_task":task,"error_log":output}, f)
self.resonance.reinforce(task.get('intent','unknown'), success=False)
print(f"[!] FAILURE LOGGED. Auto-Debug initialized: {path}")
def start(self):
print("[+] FSI Kernel Daemon Active (Self-Healing + Resonance Enabled).")
while True:
if os.path.exists(self.task_file):
with open(self.task_file, 'r') as f:
try: task = json.load(f)
except json.JSONDecodeError:
print("[!] Error reading task file. Retrying...")
time.sleep(1)
continue
intent = task.get('intent')
try:
if intent == 'scaffold':
res = self.kernel.scaffold_module(task.get('module_name'))
else:
res = self.kernel.write_code(task.get('file'), task.get('code'))
target = self.root if intent == 'scaffold' else os.path.dirname(
os.path.join(self.root, task.get('file', '')))
success, output = KernelValidator.run_tests(target)
if success:
self.ledger.update_state(intent, "Completed")
self.resonance.reinforce(intent, success=True)
try:
from src.brain.pattern_library import PatternLibrary
PatternLibrary().store(intent, task.get('code') or intent, task.get('file'))
except Exception as e:
print(f"[DAEMON] Pattern learning skipped: {e}")
print(f"[SUCCESS] Action: {intent}. Resonance reinforced.")
else:
self.handle_failure(task, output)
self.ledger.update_state(intent, f"Failed: {output[:50]}...")
except Exception as e:
print(f"[CRITICAL ERROR] {e}")
os.remove(self.task_file)
time.sleep(1)
if __name__ == "__main__":
KernelDaemon(os.getcwd()).start()
|