edge_registry

app/services/edge_registry.py

@@ -1,8 +1,8 @@
 """
 FILE: app/services/edge_registry.py
 DESCRIPTION: Single Source of Truth für Kanten-Typen mit dynamischem Reload.
-             WP-22: Transparente Status-Meldungen für Dev-Umgebungen.
+             WP-22: Transparente Status-Meldungen & Robuste Pfad-Auflösung.
-VERSION: 0.7.2 (Fix: Restore Console Visibility & Entry Counts)
+VERSION: 0.7.3 (Fix: Path Normalization & Env Priority)
 """
 import re
 import os
@@ -17,7 +17,6 @@ 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):
@@ -31,10 +30,16 @@ class EdgeRegistry:
             return
             
         settings = get_settings()
+        # Hole ENV-Werte und entferne potenzielle Anführungszeichen
         env_vocab_path = os.getenv("MINDNET_VOCAB_PATH")
+        if env_vocab_path:
+            env_vocab_path = env_vocab_path.strip('"').strip("'")
+            
         env_vault_root = os.getenv("MINDNET_VAULT_ROOT") or getattr(settings, "MINDNET_VAULT_ROOT", "./vault")
+        if env_vault_root:
+            env_vault_root = env_vault_root.strip('"').strip("'")
         
-        # Pfad-Priorität: 1. ENV -> 2. _system/dictionary -> 3. 01_User_Manual
+        # Pfad-Priorität: 1. Direkter Pfad (ENV) -> 2. Vault-Struktur
         if env_vocab_path:
             self.full_vocab_path = os.path.abspath(env_vocab_path)
         else:
@@ -56,7 +61,6 @@ class EdgeRegistry:
         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
@@ -73,11 +77,10 @@ class EdgeRegistry:
             self._last_mtime = current_mtime
 
     def _load_vocabulary(self):
-        """Parst das Wörterbuch und meldet die Anzahl der gelesenen Einträge."""
+        """Parst das Wörterbuch[cite: 25]."""
         self.canonical_map.clear()
         self.valid_types.clear()
         
-        # Regex deckt | **canonical** | Aliase | ab
         pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|")
 
         try:
@@ -96,18 +99,14 @@ 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
             
-            # 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:
             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."""
@@ -117,25 +116,21 @@ class EdgeRegistry:
         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."""
+        """JSONL-Logging[cite: 1]."""
         try:
             os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True)
             entry = {