mindnet/app/services/edge_registry.py

"""
FILE: app/services/edge_registry.py
DESCRIPTION: Single Source of Truth für Kanten-Typen mit dynamischem Reload.
             WP-15b: Erweiterte Provenance-Prüfung für die Candidate-Validation.
             Sichert die Graph-Integrität durch strikte Trennung von System- und Inhaltskanten.
             WP-22: Fix für absolute Pfade außerhalb des Vaults (Prod-Dictionary).
             WP-20: Synchronisation mit zentralen Settings (v0.6.2).
VERSION: 0.8.0
STATUS: Active
DEPENDENCIES: re, os, json, logging, time, app.config
LAST_ANALYSIS: 2025-12-26
"""
import re
import os
import json
import logging
import time
from typing import Dict, Optional, Set, Tuple

from app.config import get_settings

logger = logging.getLogger(__name__)

class EdgeRegistry:
    """
    Zentraler Verwalter für das Kanten-Vokabular.
    Implementiert das Singleton-Pattern für konsistente Validierung über alle Services.
    """
    _instance = None
    # System-Kanten, die nicht durch User oder KI gesetzt werden dürfen
    FORBIDDEN_SYSTEM_EDGES = {"next", "prev", "belongs_to"}

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super(EdgeRegistry, cls).__new__(cls)
            cls._instance.initialized = False
        return cls._instance

    def __init__(self):
        if self.initialized: 
            return
            
        settings = get_settings()
        
        # 1. Pfad aus den zentralen Settings laden (WP-20 Synchronisation)
        # Priorisiert den Pfad aus der .env / config.py (v0.6.2)
        self.full_vocab_path = os.path.abspath(settings.MINDNET_VOCAB_PATH)
        
        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 Ladevorgang
        logger.info(f">>> [EDGE-REGISTRY] Initializing with Path: {self.full_vocab_path}")
        self.ensure_latest()
        self.initialized = True

    def ensure_latest(self):
        """
        Prüft den Zeitstempel der Vokabular-Datei und lädt bei Bedarf neu.
        Verhindert Inkonsistenzen bei Laufzeit-Updates des Dictionaries.
        """
        if not os.path.exists(self.full_vocab_path):
            logger.error(f"!!! [EDGE-REGISTRY ERROR] File not found: {self.full_vocab_path} !!!")
            return

        try:
            current_mtime = os.path.getmtime(self.full_vocab_path)
            if current_mtime > self._last_mtime:
                self._load_vocabulary()
                self._last_mtime = current_mtime
        except Exception as e:
            logger.error(f"!!! [EDGE-REGISTRY] Error checking file time: {e}")

    def _load_vocabulary(self):
        """
        Parst das Markdown-Wörterbuch und baut die Canonical-Map auf.
        Erkennt Tabellen-Strukturen und extrahiert fettgedruckte System-Typen.
        """
        self.canonical_map.clear()
        self.valid_types.clear()
        
        # Regex für Tabellen-Struktur: | **Typ** | Aliase |
        pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|")

        try:
            with open(self.full_vocab_path, "r", encoding="utf-8") as f:
                c_types, c_aliases = 0, 0
                for line in f:
                    match = pattern.search(line)
                    if match:
                        canonical = match.group(1).strip().lower()
                        aliases_str = match.group(2).strip()
                        
                        self.valid_types.add(canonical)
                        self.canonical_map[canonical] = canonical
                        c_types += 1
                        
                        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, Underscores statt Leerzeichen
                                clean_alias = alias.replace("`", "").lower().strip().replace(" ", "_")
                                self.canonical_map[clean_alias] = canonical
                                c_aliases += 1
            
            logger.info(f"=== [EDGE-REGISTRY SUCCESS] Loaded {c_types} Canonical Types and {c_aliases} Aliases ===")

        except Exception as e:
            logger.error(f"!!! [EDGE-REGISTRY FATAL] Error reading file: {e} !!!")

    def resolve(self, edge_type: str, provenance: str = "explicit", context: dict = None) -> str:
        """
        WP-15b: Validiert einen Kanten-Typ gegen das Vokabular und prüft Berechtigungen.
        Sichert, dass nur strukturelle Prozesse System-Kanten setzen dürfen.
        """
        self.ensure_latest()
        if not edge_type: 
            return "related_to"
        
        # Normalisierung des Typs
        clean_type = edge_type.lower().strip().replace(" ", "_").replace("-", "_")
        ctx = context or {}

        # WP-15b: System-Kanten dürfen weder manuell noch durch KI/Vererbung gesetzt werden.
        # Nur Provenienz 'structure' (interne Prozesse) ist autorisiert.
        # Wir blockieren hier alle Provenienzen außer 'structure'.
        restricted_provenance = ["explicit", "semantic_ai", "inherited", "global_pool", "rule"]
        if provenance in restricted_provenance and clean_type in self.FORBIDDEN_SYSTEM_EDGES:
            self._log_issue(clean_type, f"forbidden_usage_by_{provenance}", ctx)
            return "related_to"

        # System-Kanten sind NUR bei struktureller Provenienz erlaubt
        if provenance == "structure" and clean_type in self.FORBIDDEN_SYSTEM_EDGES:
            return clean_type

        # Mapping auf kanonischen Namen (Alias-Auflösung)
        if clean_type in self.canonical_map:
            return self.canonical_map[clean_type]
        
        # Fallback und Logging unbekannter Typen für Admin-Review
        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 die Vokabular-Optimierung."""
        try:
            os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True)
            entry = {
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
                "edge_type": edge_type,
                "error": error_kind,
                "file": ctx.get("file", "unknown"),
                "line": ctx.get("line", "unknown"),
                "note_id": ctx.get("note_id", "unknown"),
                "provenance": ctx.get("provenance", "unknown")
            }
            with open(self.unknown_log_path, "a", encoding="utf-8") as f:
                f.write(json.dumps(entry) + "\n")
        except Exception: 
            pass

# Singleton Export für systemweiten Zugriff
registry = EdgeRegistry()