diff --git a/app/services/edge_registry.py b/app/services/edge_registry.py index c22bc0b..2859baa 100644 --- a/app/services/edge_registry.py +++ b/app/services/edge_registry.py @@ -1,7 +1,8 @@ """ FILE: app/services/edge_registry.py DESCRIPTION: Single Source of Truth für Kanten-Typen mit dynamischem Reload. -VERSION: 0.7.1 (Fix: Silent Path Error & Path Alignment) + WP-22: Transparente Status-Meldungen für Dev-Umgebungen. +VERSION: 0.7.2 (Fix: Restore Console Visibility & Entry Counts) """ import re import os @@ -16,6 +17,7 @@ logger = logging.getLogger(__name__) class EdgeRegistry: _instance = None + # System-Kanten, die NIEMALS manuell im Markdown stehen dürfen FORBIDDEN_SYSTEM_EDGES = {"next", "prev", "belongs_to"} def __new__(cls, *args, **kwargs): @@ -32,11 +34,10 @@ class EdgeRegistry: env_vocab_path = os.getenv("MINDNET_VOCAB_PATH") env_vault_root = os.getenv("MINDNET_VAULT_ROOT") or getattr(settings, "MINDNET_VAULT_ROOT", "./vault") - # WP-22 FIX: Pfad-Priorität angepasst auf _system/dictionary (Architektur-Standard) + # Pfad-Priorität: 1. ENV -> 2. _system/dictionary -> 3. 01_User_Manual if env_vocab_path: self.full_vocab_path = os.path.abspath(env_vocab_path) else: - # Suche an zwei Orten: Erst System-Dir, dann User-Manual possible_paths = [ os.path.join(env_vault_root, "_system", "dictionary", "edge_vocabulary.md"), os.path.join(env_vault_root, "01_User_Manual", "01_edge_vocabulary.md") @@ -48,21 +49,22 @@ class EdgeRegistry: break if not self.full_vocab_path: - self.full_vocab_path = os.path.abspath(possible_paths[0]) # Fallback + self.full_vocab_path = os.path.abspath(possible_paths[0]) self.unknown_log_path = "data/logs/unknown_edges.jsonl" self.canonical_map: Dict[str, str] = {} self.valid_types: Set[str] = set() self._last_mtime = 0.0 + # Initialer Lade-Versuch mit Konsolen-Feedback + print(f"\n>>> [EDGE-REGISTRY] Initializing with Path: {self.full_vocab_path}", flush=True) self.ensure_latest() self.initialized = True def ensure_latest(self): - """Prüft den Zeitstempel und meldet Fehler nun explizit.""" + """Prüft den Zeitstempel und lädt bei Bedarf neu.""" if not os.path.exists(self.full_vocab_path): - # WP-22 FIX: Fehlermeldung statt silent return - logger.error(f"❌ EdgeRegistry: Vocabulary file NOT FOUND at {self.full_vocab_path}") + print(f"!!! [EDGE-REGISTRY ERROR] File not found: {self.full_vocab_path} !!!", flush=True) return current_mtime = os.path.getmtime(self.full_vocab_path) @@ -71,11 +73,11 @@ class EdgeRegistry: self._last_mtime = current_mtime def _load_vocabulary(self): - """Parst das Wörterbuch mit verbessertem Logging.""" + """Parst das Wörterbuch und meldet die Anzahl der gelesenen Einträge.""" self.canonical_map.clear() self.valid_types.clear() - # Regex deckt **typ** und **`typ`** ab + # Regex deckt | **canonical** | Aliase | ab pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|") try: @@ -94,35 +96,46 @@ class EdgeRegistry: if aliases_str and "Kein Alias" not in aliases_str: aliases = [a.strip() for a in aliases_str.split(",") if a.strip()] for alias in aliases: + # Normalisierung: Kleinschreibung und Unterstriche clean_alias = alias.replace("`", "").lower().strip().replace(" ", "_") self.canonical_map[clean_alias] = canonical c_aliases += 1 - logger.info(f"✅ EdgeRegistry reloaded: {c_types} types and {c_aliases} aliases from {self.full_vocab_path}") + # Erfolgskontrolle für das Dev-Terminal + print(f"=== [EDGE-REGISTRY SUCCESS] Loaded {c_types} Canonical Types and {c_aliases} Aliases ===", flush=True) + logger.info(f"Registry reloaded from {self.full_vocab_path}") + except Exception as e: - logger.error(f"❌ EdgeRegistry: Error reading vocabulary: {e}") + print(f"!!! [EDGE-REGISTRY FATAL] Error reading file: {e} !!!", flush=True) + logger.error(f"Error reading vocabulary: {e}") def resolve(self, edge_type: str, provenance: str = "explicit", context: dict = None) -> str: + """Validierung mit Fundort-Logging.""" self.ensure_latest() if not edge_type: return "related_to" clean_type = edge_type.lower().strip().replace(" ", "_").replace("-", "_") ctx = context or {} + # 1. Schutz der Systemkanten (Verbot für manuelle Nutzung) if provenance == "explicit" and clean_type in self.FORBIDDEN_SYSTEM_EDGES: self._log_issue(clean_type, "forbidden_system_usage", ctx) return "related_to" + # 2. Akzeptanz interner Strukturkanten if provenance == "structure" and clean_type in self.FORBIDDEN_SYSTEM_EDGES: return clean_type + # 3. Mapping via Wörterbuch if clean_type in self.canonical_map: return self.canonical_map[clean_type] + # 4. Unbekannte Kante self._log_issue(clean_type, "unknown_type", ctx) return clean_type def _log_issue(self, edge_type: str, error_kind: str, ctx: dict): + """Detailliertes JSONL-Logging für Debugging.""" try: os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True) entry = {