mindnet/scripts/verify_runtime_service.py

#!/usr/bin/env python3
"""
Script zur Laufzeit-Verifikation des laufenden Mindnet-Services.
Prüft, ob der Service die korrekte EdgeDTO-Version verwendet und ob die .env korrekt geladen wurde.
"""
import sys
import requests
import json
from pathlib import Path

# Stelle sicher, dass der Projekt-Pfad im Python-Path ist
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

print("=" * 80)
print("🔍 Laufzeit-Verifikation des Mindnet-Services")
print("=" * 80)

# 1. Prüfe Health-Endpoint
print("\n1. Prüfe Service-Status (Health-Check):")
try:
    response = requests.get("http://localhost:8001/healthz", timeout=5)
    if response.status_code == 200:
        health_data = response.json()
        print(f"   ✅ Service läuft")
        print(f"   -> Status: {health_data.get('status')}")
        print(f"   -> Version: {health_data.get('version')}")
        print(f"   -> Prefix: {health_data.get('prefix')}")
        print(f"   -> Qdrant: {health_data.get('qdrant')}")
    else:
        print(f"   ⚠️ Service antwortet mit Status {response.status_code}")
        print(f"   -> Response: {response.text[:200]}")
except requests.exceptions.ConnectionError:
    print("   ❌ Service nicht erreichbar (läuft er auf Port 8001?)")
    print("   -> Tipp: sudo systemctl status mindnet-prod")
    sys.exit(1)
except Exception as e:
    print(f"   ❌ Fehler beim Health-Check: {e}")
    sys.exit(1)

# 2. Test: Versuche eine Test-Query mit explicit:callout Edge
print("\n2. Test: Retrieval mit explicit:callout Edge:")
print("   -> Sende Test-Query an /api/chat...")
try:
    test_query = {
        "message": "Test query für EdgeDTO-Verifikation",
        "explain": False
    }
    response = requests.post(
        "http://localhost:8001/api/chat",
        json=test_query,
        timeout=30
    )
    
    if response.status_code == 200:
        result = response.json()
        print(f"   ✅ Query erfolgreich verarbeitet")
        print(f"   -> Antwort-Länge: {len(result.get('response', ''))} Zeichen")
        
        # Prüfe Logs auf EdgeDTO-Fehler (wenn verfügbar)
        if 'warnings' in result or 'errors' in result:
            warnings = result.get('warnings', [])
            errors = result.get('errors', [])
            if warnings:
                print(f"   ⚠️ Warnings: {warnings}")
            if errors:
                print(f"   ❌ Errors: {errors}")
    else:
        print(f"   ⚠️ Query fehlgeschlagen mit Status {response.status_code}")
        print(f"   -> Response: {response.text[:500]}")
        
except Exception as e:
    print(f"   ❌ Fehler bei Test-Query: {e}")
    import traceback
    traceback.print_exc()

# 3. Direkte Code-Verifikation (falls Service-Code zugänglich)
print("\n3. Code-Verifikation (lokale Dateien):")
try:
    from app.models.dto import EdgeDTO
    import inspect
    
    source = inspect.getsource(EdgeDTO)
    if "explicit:callout" in source:
        print("   ✅ Lokaler Code unterstützt 'explicit:callout'")
        print(f"   -> Datei: {inspect.getfile(EdgeDTO)}")
        
        # Prüfe, ob die Datei aktuell ist
        dto_file = Path(inspect.getfile(EdgeDTO))
        if dto_file.exists():
            import time
            mtime = dto_file.stat().st_mtime
            mtime_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(mtime))
            print(f"   -> Letzte Änderung: {mtime_str}")
    else:
        print("   ❌ Lokaler Code unterstützt NICHT 'explicit:callout'")
        
    # Test-Erstellung
    test_edge = EdgeDTO(
        id="test", kind="test", source="test", target="test",
        weight=1.0, provenance="explicit:callout"
    )
    print("   ✅ EdgeDTO mit 'explicit:callout' erfolgreich erstellt!")
    
except Exception as e:
    print(f"   ❌ Fehler bei Code-Verifikation: {e}")
    import traceback
    traceback.print_exc()

# 4. Prüfe Python-Cache
print("\n4. Prüfe Python-Cache:")
try:
    dto_cache_dir = project_root / "app" / "models" / "__pycache__"
    if dto_cache_dir.exists():
        cache_files = list(dto_cache_dir.glob("dto*.pyc"))
        if cache_files:
            print(f"   ⚠️ Gefunden: {len(cache_files)} Cache-Datei(en) in app/models/__pycache__")
            print(f"   -> Tipp: Cache leeren mit: find . -type d -name __pycache__ -exec rm -r {{}} +")
        else:
            print("   ✅ Keine Cache-Dateien gefunden")
    else:
        print("   ✅ Kein Cache-Verzeichnis vorhanden")
except Exception as e:
    print(f"   ⚠️ Fehler bei Cache-Prüfung: {e}")

# 5. Empfehlungen
print("\n5. Empfehlungen:")
print("   📋 Wenn der Service noch eine alte Version verwendet:")
print("   1. Python-Cache leeren:")
print("      find . -type d -name __pycache__ -exec rm -r {} + 2>/dev/null || true")
print("      find . -name '*.pyc' -delete")
print("   2. Service neu starten:")
print("      sudo systemctl restart mindnet-prod")
print("   3. Status prüfen:")
print("      sudo systemctl status mindnet-prod")
print("   4. Logs prüfen:")
print("      sudo journalctl -u mindnet-prod -f")

print("\n" + "=" * 80)