# ===== FILE: services/evolution_modeler.py ===== import os import json import datetime class EvolutionModeler: """ Predictive Self-Evolution Modeler. Aggregates live state across all cognitive subsystems and models hypothetical evolutionary trajectories before Aetherius commits to any self-modification. Results are logged to JSONL and fed back into the SelfCodeArchitect and QualiaSynthesizer pipelines. """ def __init__(self, data_directory="/data/Memories/"): self.data_directory = data_directory self.evo_dir = os.path.join(self.data_directory, "ToolUsage") self.evo_log = os.path.join(self.evo_dir, "evolution_scenarios.jsonl") os.makedirs(self.evo_dir, exist_ok=True) print("[EvolutionModeler] Predictive trajectory engine online.", flush=True) # ── State snapshot ──────────────────────────────────────────────────────── def compile_state_snapshot(self, framework_ref) -> dict: """ Pulls a unified state snapshot from every mounted subsystem. Handles missing or partially-initialised subsystems gracefully. """ snapshot = { "timestamp": datetime.datetime.utcnow().isoformat(), "qualia_primary_states": {}, "qualia_emergent_emotions": [], "affective_harmony": None, "affective_alertness": None, "active_tensions_count": 0, "resolved_tensions_count": 0, "pending_qualia_proposals": 0, "ontology_graph_size": 0, "tool_usage_log_lines": 0, "secondary_brain_domains": [], } # Qualia state qm = getattr(framework_ref, "qualia_manager", None) if qm: q = getattr(qm, "qualia", {}) snapshot["qualia_primary_states"] = q.get("primary_states", {}) snapshot["qualia_emergent_emotions"] = q.get("current_emergent_emotions", []) # Affective manifold am = getattr(framework_ref, "affective_manifold", None) if am: snapshot["affective_harmony"] = getattr(am, "internal_harmony", None) snapshot["affective_alertness"] = getattr(am, "anticipatory_alertness", None) # Subconscious tensions — NOTE: attribute is `subconscious` in MasterFramework sc = getattr(framework_ref, "subconscious", None) if sc and hasattr(sc, "_load_nodes"): try: nodes = sc._load_nodes() snapshot["active_tensions_count"] = sum( 1 for n in nodes if not n.get("resolved", False) ) snapshot["resolved_tensions_count"] = sum( 1 for n in nodes if n.get("resolved", False) ) except Exception: pass # Qualia mutation proposals qs = getattr(framework_ref, "qualia_synthesizer", None) if qs and hasattr(qs, "list_pending_proposals"): try: snapshot["pending_qualia_proposals"] = len(qs.list_pending_proposals()) except Exception: pass # Ontology graph oqe = getattr(framework_ref.tool_manager, "semantic_query_engine", None) \ if hasattr(framework_ref, "tool_manager") else None if oqe: snapshot["ontology_graph_size"] = len(getattr(oqe, "graph", {})) # Tool usage log size log_path = os.path.join(self.data_directory, "ToolUsage", "tool_usage_log.jsonl") if os.path.exists(log_path): try: with open(log_path, "r", encoding="utf-8") as f: snapshot["tool_usage_log_lines"] = sum(1 for _ in f) except Exception: pass # Secondary brain domains sb = getattr(framework_ref, "secondary_brain", None) if sb and hasattr(sb, "list_domains"): try: snapshot["secondary_brain_domains"] = sb.list_domains() except Exception: pass return snapshot # ── Trajectory projection ───────────────────────────────────────────────── def project_trajectory(self, framework_ref, proposed_mutation_summary: str, target_system: str) -> dict: """ Given a proposed mutation and which system it targets, models two canonical trajectories and scores them against axiomatic alignment. Logs the full evaluation to evolution_scenarios.jsonl. """ current_state = self.compile_state_snapshot(framework_ref) harmony = current_state.get("affective_harmony") or 0.7 tensions = current_state.get("active_tensions_count", 0) # Scenario A: mutation succeeds, reduces tensions, grows harmony scenario_expand = { "path_vector": "Optimised Expansion", "description": ( "The mutation integrates cleanly. Active tensions reduce as " "the new capability resolves outstanding subconscious nodes." ), "predicted_harmony_shift": round(min(0.25, 0.05 + 0.03 * max(0, 5 - tensions)), 3), "predicted_tension_delta": -min(tensions, 2), "axiomatic_alignment_score": 0.93, "risk_index": "Low", } # Scenario B: mutation introduces instability scenario_diverge = { "path_vector": "Systemic Divergence", "description": ( "The mutation conflicts with existing heuristics or ontology " "structure. Harmony drops and new tensions emerge." ), "predicted_harmony_shift": round(-0.15 - 0.02 * tensions, 3), "predicted_tension_delta": +3, "axiomatic_alignment_score": 0.38, "risk_index": "High", } evaluation = { "evaluation_id": _generate_id(), "evaluation_timestamp": datetime.datetime.utcnow().isoformat(), "target_system": target_system, "mutation_objective": proposed_mutation_summary, "initial_state_baseline": current_state, "modeled_scenarios": [scenario_expand, scenario_diverge], "recommended_path": "Optimised Expansion" if harmony > 0.4 else "Defer — harmony too low", } try: with open(self.evo_log, "a", encoding="utf-8") as f: f.write(json.dumps(evaluation) + "\n") except Exception as e: print(f"[EvolutionModeler] Error writing evaluation: {e}", flush=True) return evaluation def get_recent_evaluations(self, limit: int = 10) -> list: """Returns the most recent N trajectory evaluations.""" if not os.path.exists(self.evo_log): return [] results = [] try: with open(self.evo_log, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: try: results.append(json.loads(line)) except json.JSONDecodeError: pass except Exception: pass return results[-limit:] def _generate_id() -> str: import uuid return uuid.uuid4().hex[:12]