""" ComplexityReasoner — Vitalis FSI Assesses how hard a problem is BEFORE attempting it. Allocates cognitive resources accordingly. Prevents wasted cycles on problems beyond current capability. Prevents under-allocation on trivial problems. Complexity dimensions: - Structural: how many components does this problem have? - Novelty: how far is this from known patterns? - Depth: how many reasoning steps are required? - Ambiguity: how many valid interpretations exist? """ import numpy as np import os import json import time from vitalis_ide.math_core.kernel import VitalisKernel from src.cognition.abstraction import AbstractionEngine from src.hippocampus import Hippocampus class ComplexityReasoner: # Complexity tier thresholds TRIVIAL_THRESHOLD = 0.25 SIMPLE_THRESHOLD = 0.45 MODERATE_THRESHOLD = 0.65 COMPLEX_THRESHOLD = 0.80 # Resource allocation per tier RESOURCE_MAP = { "TRIVIAL": {"cycles": 1, "abstraction_depth": 1, "analogy_search": False}, "SIMPLE": {"cycles": 2, "abstraction_depth": 2, "analogy_search": False}, "MODERATE": {"cycles": 4, "abstraction_depth": 3, "analogy_search": True}, "COMPLEX": {"cycles": 8, "abstraction_depth": 4, "analogy_search": True}, "FRONTIER": {"cycles": 16, "abstraction_depth": 5, "analogy_search": True}, } def __init__(self): self.kernel = VitalisKernel() self.abstraction = AbstractionEngine() self.hippocampus = Hippocampus() self.path = os.path.expanduser( "~/.vitalis_workspace/complexity_log.json" ) self._log = self._load_log() self._history = [] def _load_log(self) -> list: if os.path.exists(self.path): with open(self.path) as f: return json.load(f) return [] def _save_log(self): os.makedirs(os.path.dirname(self.path), exist_ok=True) with open(self.path, "w") as f: json.dump(self._log[-1000:], f, indent=2) # ------------------------------------------------------------------ # Core assessment # ------------------------------------------------------------------ def assess(self, intent: str, context: dict = None) -> dict: """ Full complexity assessment for an intent. Returns complexity score, tier, and resource allocation. """ context = context or {} tokens = intent.lower().split() vec = self.kernel.vectorize_tokens(tokens, positional=False) # 1. Structural complexity — token count and unique concepts structural = self._structural_score(tokens) # 2. Novelty — distance from known patterns novelty = self._novelty_score(vec) # 3. Depth — estimated reasoning steps needed depth = self._depth_score(intent, tokens) # 4. Ambiguity — spread across abstraction space ambiguity = self._ambiguity_score(vec) # Weighted composite score = ( structural * 0.20 + novelty * 0.35 + depth * 0.25 + ambiguity * 0.20 ) score = float(np.clip(score, 0.0, 1.0)) tier = self._tier(score) resources = self.RESOURCE_MAP[tier].copy() result = { "intent": intent, "score": round(score, 4), "tier": tier, "dimensions": { "structural": round(structural, 4), "novelty": round(novelty, 4), "depth": round(depth, 4), "ambiguity": round(ambiguity, 4), }, "resources": resources, "timestamp": time.time(), } self._log.append(result) self._history.append(score) if len(self._history) > 100: self._history.pop(0) self._save_log() return result # ------------------------------------------------------------------ # Dimension calculators # ------------------------------------------------------------------ def _structural_score(self, tokens: list) -> float: """More tokens + unique concepts = higher structural complexity.""" n = len(tokens) unique = len(set(tokens)) # Normalise: 10 tokens = moderate complexity return float(np.clip((n / 10.0) * 0.5 + (unique / n if n > 0 else 0) * 0.5, 0, 1)) def _novelty_score(self, vec: np.ndarray) -> float: """ How far is this from anything the system has seen before? High novelty = far from known abstractions. """ candidates = self.abstraction.query_abstractions(vec, top_k=1) if not candidates: return 1.0 # completely novel best_sim = candidates[0][0] return float(np.clip(1.0 - best_sim, 0.0, 1.0)) def _depth_score(self, intent: str, tokens: list) -> float: """ Estimate reasoning depth from linguistic markers. Multi-step intents score higher. """ depth_markers = { "high": ["analyze", "verify", "optimize", "refactor", "debug", "compare", "evaluate", "synthesize"], "medium": ["write", "scaffold", "create", "build", "generate", "implement"], "low": ["run", "check", "show", "list", "get"], } for token in tokens: if token in depth_markers["high"]: return 0.8 if token in depth_markers["medium"]: return 0.5 if token in depth_markers["low"]: return 0.2 # Connectives suggest multi-step reasoning if any(w in intent.lower() for w in ["and then", "after", "before", "while"]): return 0.9 return 0.4 # default moderate def _ambiguity_score(self, vec: np.ndarray) -> float: """ How spread are the top matches in abstraction space? High spread = high ambiguity = harder problem. """ candidates = self.abstraction.query_abstractions(vec, top_k=3) if len(candidates) < 2: return 0.5 scores = [c[0] for c in candidates] spread = float(np.std(scores)) return float(np.clip(spread * 4.0, 0.0, 1.0)) def _tier(self, score: float) -> str: if score < self.TRIVIAL_THRESHOLD: return "TRIVIAL" elif score < self.SIMPLE_THRESHOLD: return "SIMPLE" elif score < self.MODERATE_THRESHOLD: return "MODERATE" elif score < self.COMPLEX_THRESHOLD: return "COMPLEX" else: return "FRONTIER" # ------------------------------------------------------------------ # Trend analysis # ------------------------------------------------------------------ def complexity_trend(self) -> dict: """Is the system tackling harder or easier problems over time?""" if len(self._history) < 5: return {"status": "Insufficient data"} recent = float(np.mean(self._history[-5:])) baseline = float(np.mean(self._history)) trend = "increasing" if recent > baseline + 0.05 else \ "decreasing" if recent < baseline - 0.05 else "stable" return { "recent_avg": round(recent, 4), "baseline": round(baseline, 4), "trend": trend, "sample_size": len(self._history), } def report(self) -> dict: if not self._log: return {"status": "No assessments yet"} tiers = {} for e in self._log: t = e.get("tier", "UNKNOWN") tiers[t] = tiers.get(t, 0) + 1 return { "total_assessments": len(self._log), "tier_distribution": tiers, "avg_complexity": round(float(np.mean([e["score"] for e in self._log])), 4), "trend": self.complexity_trend(), }