MCP Server Refactoring Plan

Version: 1.2
Erstellt: 2025-12-28
Aktualisiert: 2025-12-28 (nach Supervision Runde 2)
Status: Final
Task: #507

1. Executive Summary

Ist-Zustand

• 5 MCP-Server: mcp-db, mcp-tasks, mcp-contracts, mcp-docs, mcp-code
• 91 Python-Dateien (ohne venv)
• 9.248 LOC gesamt
• Signifikante Code-Duplikation (~25% redundanter Code)
• Inkonsistente Patterns zwischen Servern

Ziel

Konsolidierung zu einer modularen, wartbaren Architektur unter Einhaltung von:

• DRY (Don't Repeat Yourself)
• SRP (Single Responsibility Principle)
• SOLID Principles
• Clean Architecture
• KISS (Keep It Simple, Stupid)
• YAGNI (You Aren't Gonna Need It)

Erwartete Verbesserungen

• ~30% weniger Code durch Elimination von Duplikaten
• Einheitliche Patterns für alle Server
• Verbesserte Wartbarkeit durch Modularisierung
• Schnellere Feature-Entwicklung durch shared Components

2. Analyse der Code-Duplikation

2.1 Kritische Duplikate (Sofort beheben)

db_connection.py (4 Dateien, ~240 LOC dupliziert)

Problem: 3 von 4 Dateien sind nahezu identisch (mcp-tasks, mcp-contracts, mcp-docs).

Lösung: Zentralisieren in shared/infrastructure/simple_db_connection.py

WICHTIG: Diese Klasse heißt bewusst SimpleDbConnection, nicht DatabaseConnection. Sie bietet kein Pooling und ist nicht für High-Throughput geeignet. Für mcp-db (100+ Queries/Minute) bleibt die eigene Pooling-Implementierung.

# shared/infrastructure/simple_db_connection.py
"""
Einfache DB-Verbindung ohne Pooling.

NICHT geeignet für:
- High-Throughput (> 50 Queries/Minute)
- Connection Pooling Requirements
- mcp-db Server

Für diese Fälle: Eigene Implementierung mit mysql.connector.pooling
"""
from contextlib import contextmanager
from typing import Generator

import pymysql
from pymysql.connections import Connection

from shared.config_base import AppDatabaseConfig, LogDatabaseConfig


class SimpleDbConnection:
    """
    Einfache Datenbankverbindung ohne Pooling.
    
    Für Server mit geringem Query-Volumen (< 50/Minute).
    """
    
    @classmethod
    @contextmanager
    def get_connection(
        cls, 
        config: AppDatabaseConfig,
        database: str | None = None,
        autocommit: bool = False
    ) -> Generator[Connection, None, None]:
        """
        Context Manager für App-DB Connection.
        
        ACHTUNG: Keine weiteren Parameter hinzufügen!
        Bei Bedarf für mehr Logik: Neue Methode erstellen.
        
        Args:
            config: Server-Konfiguration mit DB-Credentials
            database: Optional - überschreibt config.DB_NAME
            autocommit: True für Logging, False für Transaktionen
        """
        conn = None
        db_name = database or getattr(config, 'DB_NAME', None)
        
        try:
            conn = pymysql.connect(
                host=config.DB_HOST,
                port=getattr(config, 'DB_PORT', 3306),
                user=config.DB_USER,
                password=config.DB_PASSWORD,
                database=db_name,
                charset="utf8mb4",
                cursorclass=pymysql.cursors.DictCursor,
                autocommit=autocommit,
            )
            yield conn
            if not autocommit:
                conn.commit()
        except Exception:
            if conn and not autocommit:
                conn.rollback()
            raise
        finally:
            if conn:
                conn.close()
    
    @classmethod
    @contextmanager
    def get_log_connection(
        cls, 
        config: LogDatabaseConfig
    ) -> Generator[Connection, None, None]:
        """Separate Verbindung für Logging (autocommit=True)."""
        conn = None
        try:
            conn = pymysql.connect(
                host=config.LOG_DB_HOST,
                user=config.LOG_DB_USER,
                password=config.LOG_DB_PASSWORD,
                database=config.LOG_DB_NAME,
                charset="utf8mb4",
                cursorclass=pymysql.cursors.DictCursor,
                autocommit=True,
            )
            yield conn
        finally:
            if conn:
                conn.close()

protokoll_logger.py (4 Dateien, ~240 LOC dupliziert)

Lösung: Zentralisieren mit strikter Clean Architecture Trennung

# shared/domain/log_entry.py
"""
Domain Entity für Log-Einträge.

REINE DOMAIN - keine Infrastructure-Abhängigkeiten!
"""
from dataclasses import dataclass
from datetime import datetime
from typing import Optional


@dataclass(frozen=True)
class LogEntry:
    """Standardisierter Log-Eintrag für alle MCP-Server."""
    timestamp: datetime
    client_name: str
    request: str
    status: str
    duration_ms: int
    error_message: Optional[str] = None
    tool_name: Optional[str] = None
    context_id: Optional[int] = None  # task_id, contract_id, etc.

# shared/infrastructure/protokoll_logger.py
"""
Infrastructure Logger für mcp_log Tabelle.

Importiert LogEntry aus Domain - definiert ihn NICHT selbst.
"""
import logging
import sys
from typing import TYPE_CHECKING

from shared.domain.log_entry import LogEntry  # Import aus Domain!
from shared.infrastructure.simple_db_connection import SimpleDbConnection

if TYPE_CHECKING:
    from shared.config_base import LogDatabaseConfig


class ProtokollLogger:
    """Fail-Safe Logger für mcp_log Tabelle."""
    
    def __init__(self, client_name: str, config: "LogDatabaseConfig"):
        self.client_name = client_name
        self.config = config
        self._logger = logging.getLogger(f"mcp.{client_name}")
    
    def log(self, entry: LogEntry) -> None:
        """Schreibt Log-Eintrag. Fehler gehen nur zu stderr."""
        try:
            with SimpleDbConnection.get_log_connection(self.config) as conn:
                with conn.cursor() as cursor:
                    request_str = self._format_request(entry)
                    cursor.execute(
                        """INSERT INTO mcp_log
                           (timestamp, client_name, request, status, duration_ms, error_message)
                           VALUES (%s, %s, %s, %s, %s, %s)""",
                        (
                            entry.timestamp,
                            self.client_name,
                            request_str[:500],
                            entry.status,
                            entry.duration_ms,
                            entry.error_message[:500] if entry.error_message else None,
                        )
                    )
        except Exception as e:
            print(f"CRITICAL: {self.client_name} log failed: {e}", file=sys.stderr)
    
    def _format_request(self, entry: LogEntry) -> str:
        parts = []
        if entry.tool_name:
            parts.append(f"[{entry.tool_name}]")
        if entry.context_id:
            parts.append(f"id={entry.context_id}")
        parts.append(entry.request[:400] if entry.request else "")
        return " ".join(parts)


# === Registry mit Test-Support ===

_logger_instances: dict[str, ProtokollLogger] = {}


def get_logger(client_name: str, config: "LogDatabaseConfig") -> ProtokollLogger:
    """Singleton-Factory für Logger."""
    if client_name not in _logger_instances:
        _logger_instances[client_name] = ProtokollLogger(client_name, config)
    return _logger_instances[client_name]


def clear_logger_registry() -> None:
    """
    Setzt die Logger-Registry zurück.
    
    NUR FÜR TESTS! Ermöglicht isolierte Test-Ausführung.
    """
    _logger_instances.clear()

config.py - Strikte Statische Konfiguration

# shared/config_base.py
"""
Basis-Konfiguration für alle MCP-Server.

WICHTIG: Diese Klasse wird NICHT instanziiert!
Alle Zugriffe erfolgen über Klassenattribute.

Falsch: config = BaseConfig()
Richtig: host = BaseConfig.DB_HOST
"""
import os
from typing import ClassVar, Protocol, runtime_checkable


@runtime_checkable
class AppDatabaseConfig(Protocol):
    """Protocol für App-Datenbank Konfiguration."""
    DB_HOST: str
    DB_PORT: int
    DB_USER: str
    DB_PASSWORD: str
    DB_NAME: str


@runtime_checkable
class LogDatabaseConfig(Protocol):
    """Protocol für Log-Datenbank Konfiguration."""
    LOG_DB_HOST: str
    LOG_DB_NAME: str
    LOG_DB_USER: str
    LOG_DB_PASSWORD: str


class BaseConfig:
    """
    Statische Konfiguration - NICHT INSTANZIIEREN!
    
    Verwendung:
        from shared.config_base import BaseConfig
        host = BaseConfig.DB_HOST  # Korrekt
        
        config = BaseConfig()  # FALSCH!
    """
    
    # App-Datenbank
    DB_HOST: ClassVar[str] = os.getenv("DB_HOST", "localhost")
    DB_PORT: ClassVar[int] = int(os.getenv("DB_PORT", "3306"))
    DB_USER: ClassVar[str] = os.getenv("DB_USER", "root")
    DB_PASSWORD: ClassVar[str] = os.getenv("DB_PASSWORD", "")
    DB_NAME: ClassVar[str] = os.getenv("DB_NAME", "ki_dev")
    
    # Log-Datenbank
    LOG_DB_HOST: ClassVar[str] = os.getenv("LOG_DB_HOST", "localhost")
    LOG_DB_NAME: ClassVar[str] = os.getenv("LOG_DB_NAME", "ki_dev")
    LOG_DB_USER: ClassVar[str] = os.getenv("LOG_DB_USER", "mcp_logger")
    LOG_DB_PASSWORD: ClassVar[str] = os.getenv("LOG_DB_PASSWORD", "")
    
    def __init__(self):
        raise TypeError("BaseConfig ist statisch und darf nicht instanziiert werden")

3. SRP-Verletzungen

3.1 ContractValidator (409 Zeilen) - Pragmatische Teilzerlegung

Empfohlenes Refactoring:

validators/
├── contract_validator.py      # Orchestrierung (< 150 Zeilen)
├── scope_resolver.py          # Pfad-Auflösung + Glob (< 100 Zeilen)
└── rule_evaluator.py          # Domain Service für Regelauswertung

HINWEIS: rule_evaluator.py ist ein Domain Service, kein Repository! Es führt keine CRUD-Operationen durch, sondern Business-Logik. Keine Repository-Nomenklatur für Validatoren verwenden.

Aktuelle Verantwortlichkeiten von rule_evaluator:

1. Regel-Parsing
2. Regel-Auswertung
3. Regel-Aggregation

Status: Als Übergang akzeptabel, da reines In-Memory-Domain-Modul ohne IO.

Grenze: Sobald eine Regel externen Zustand berührt → weitere Zerlegung nötig.

4. Inkonsistenzen

4.1 Datenbank-Bibliotheken - Hybride Strategie

Trade-off explizit dokumentiert:

• pymysql ohne Pooling: ~50-100ms Connection-Overhead pro Request
• Für mcp-tasks/contracts/docs/code: Akzeptabel bei < 50 Queries/Minute
• Für mcp-db: Inakzeptabel, daher eigenes Pooling

4.2 Zwei DB-Abstraktionen (konzedierte Redundanz)

Semantische Abgrenzung:

• SimpleDbConnection = Explizit "einfach", nicht für Hochlast
• mcp-db behält eigene Implementierung mit MySQLConnectionPool

5. Ziel-Architektur

5.1 Verzeichnisstruktur (nach Package-Umbenennung)

/var/www/mcp-servers/
├── shared/                           # Gemeinsame Komponenten
│   ├── __init__.py
│   ├── config_base.py               # Statische BaseConfig + Protocols
│   ├── constants.py                 # ✓ Existiert bereits
│   ├── server_factory.py            # Server Creation Factory
│   ├── domain/
│   │   ├── __init__.py
│   │   └── log_entry.py             # LogEntry Dataclass (REINE DOMAIN)
│   └── infrastructure/
│       ├── __init__.py
│       ├── simple_db_connection.py  # Ohne Pooling, nicht für Hochlast
│       └── protokoll_logger.py      # Importiert LogEntry aus domain/
│
├── mcp_db/                          # Umbenannt von mcp-db!
│   ├── server.py
│   ├── config.py
│   ├── infrastructure/
│   │   └── db_connection.py         # EIGENE Version mit Pooling
│   └── tools/
│
├── mcp_tasks/                       # Umbenannt von mcp-tasks!
│   ├── server.py
│   ├── config.py
│   ├── domain/
│   ├── infrastructure/
│   │   └── task_repository.py
│   └── tools/
│
├── mcp_contracts/                   # Umbenannt!
│   ├── server.py
│   ├── config.py
│   ├── validators/
│   │   ├── contract_validator.py
│   │   ├── scope_resolver.py
│   │   └── rule_evaluator.py        # Domain Service, KEIN Repository!
│   └── tools/
│
├── mcp_docs/                        # Umbenannt!
│   └── ...
│
└── mcp_code/                        # Umbenannt!
    └── ...

5.2 Clean Architecture Schichten

┌─────────────────────────────────────────────────────────┐
│                    SHARED/DOMAIN                         │
│  ┌───────────────────────────────────────────────────┐  │
│  │ log_entry.py (reine Dataclass, keine Deps)       │  │
│  └───────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘
                          ▲
                          │ importiert
┌─────────────────────────────────────────────────────────┐
│                 SHARED/INFRASTRUCTURE                    │
│  ┌───────────────────────────────────────────────────┐  │
│  │ simple_db_connection.py, protokoll_logger.py     │  │
│  └───────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘
                          ▲
                          │ importiert
┌─────────────────────────────────────────────────────────┐
│                    SERVER-LAYER                          │
│  ┌─────────┐  ┌───────────┐  ┌─────────┐  ┌─────────┐  │
│  │ mcp_db  │  │ mcp_tasks │  │mcp_docs │  │mcp_code │  │
│  └─────────┘  └───────────┘  └─────────┘  └─────────┘  │
└─────────────────────────────────────────────────────────┘

6. Implementierungsplan mit Quality Gates

Phase 0: Package-Umbenennung (VORAUSSETZUNG)

Dauer: 0.5 Tage

Gate:

• [ ] Keine sys.path.insert mehr notwendig
• [ ] pyproject.toml konsistent
• [ ] Alle Server starten und antworten

ACHTUNG: Phase 0 ist nicht optional! Ohne korrekte Package-Namen ist das gesamte Refactoring auf Sand gebaut.

Phase 1: Shared Foundation

Dauer: 2 Tage

Phase 1 Abnahmekriterien:

• [ ] ruff check shared/ = 0 Fehler
• [ ] mypy shared/ = 0 Fehler
• [ ] Domain Coverage: 100%
• [ ] Infrastructure Coverage: >= 90%
• [ ] clear_logger_registry() existiert und funktioniert
• [ ] Kein sys.path.insert in shared/

Phase 2: Server Migration

Phase 2a: mcp_docs (einfachstes)

Dauer: 0.5 Tage

Gate:

• [ ] Nutzt SimpleDbConnection
• [ ] Nutzt ProtokollLogger aus shared/
• [ ] Kein sys.path.insert
• [ ] LOC-Reduktion: >= 50 Zeilen

Phase 2b-2e: Weitere Server

(analog, siehe v1.1)

Universelle Regel für alle Server:

• [ ] Kein sys.path.insert nach Migration
• [ ] Nur relative Imports oder Package-Imports

Phase 3: ContractValidator Refactoring

7. Verbote und Grenzen

7.1 Absolute Verbote nach Phase 1

7.2 Komplexitätsgrenzen

8. Qualitätssicherung

Coverage-Strategie (differenziert)

Test-Isolation

# In jedem Test-Setup:
from shared.infrastructure.protokoll_logger import clear_logger_registry

def setup_function():
    clear_logger_registry()  # Isolierte Tests

9. Metriken

Vorher (Baseline)

Ziel Nachher (mit Phase-Gates)

10. Entscheidungen (Final)

Entscheidung 1: Package-Umbenennung

Entscheidung: Phase 0 (Voraussetzung, nicht optional)

Entscheidung 2: Datenbank-Bibliothek

Entscheidung: Hybrid mit klarer Semantik

• SimpleDbConnection für Standard-Server (kein Pooling)
• Eigene Implementierung für mcp_db (mit Pooling)

Entscheidung 3: Singleton Pattern

Entscheidung: Dict-Registry mit clear_registry() für Tests

Entscheidung 4: Clean Architecture

Entscheidung: Strikte Trennung Domain/Infrastructure

• LogEntry nur in shared/domain/log_entry.py
• Infrastructure importiert aus Domain, nicht umgekehrt

Entscheidung 5: sys.path.insert

Entscheidung: Vollständig verboten nach Phase 0

Anhang A: Korrekturen nach Supervision

Runde 1 (v1.0 → v1.1)

Runde 2 (v1.1 → v1.2)

Dokument Version 1.2 - Final für Implementierung.

Claude Pre-/Post-Hook System

Erstellt: 2025-12-23 | Aktualisiert: 2025-12-27

Das Hook-System greift automatisch bei jeder Interaktion mit Claude Code ein und stellt sicher, dass alle Aktionen protokolliert, validiert und abgesichert werden.

Operationale Übersicht: Claude Hooks (Betrieb)

Hook-Typen

Konfiguration

Die Hooks werden in /root/.claude/settings.json konfiguriert.

Ablaufdiagramm: Edit/Write Operation

┌─────────────────┐
│  User-Prompt    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ UserPromptSubmit│──▶ log_to_db.py (Logging)
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   PreToolUse    │
│   (Edit/Write)  │
└────────┬────────┘
         │
    ┌────┴────┐
    │         │
    ▼         ▼
┌───────┐ ┌───────────┐ ┌────────────┐
│Block- │ │file_backup│ │log_to_db.py│
│DB.py  │ │_hook.py   │ │(Logging)   │
└───┬───┘ └─────┬─────┘ └──────┬─────┘
    │           │              │
    │      Backup erstellt     │
    │           │              │
    └───────────┴──────────────┘
                │
         ┌──────┴──────┐
         │  Tool       │
         │  Execution  │
         └──────┬──────┘
                │
                ▼
┌─────────────────┐
│  PostToolUse    │
│  (Edit/Write)   │
└────────┬────────┘
         │
    ┌────┴────┐
    │         │
    ▼         ▼
┌───────────┐ ┌────────────┐
│fix-permi- │ │log_to_db.py│
│ssions.sh  │ │(Response)  │
└─────┬─────┘ └──────┬─────┘
      │              │
      ▼              │
  Permissions        │
  korrigiert         │
      └──────────────┘

Unterkapitel

• Pre-Hook Ablauf - Was passiert VOR der Tool-Ausführung
• Post-Hook Ablauf - Was passiert NACH der Tool-Ausführung
• Hook-Scripts - Übersicht aller Scripts
• Protokollierung - Logging in ki_dev.protokoll
• Task-Completion Guard - Workflow-Validierung

Verwandte Themen

• Claude Hooks (Betrieb) - Alle Hooks mit Konfiguration
• Quality Gates - Code-Qualitätsprüfung

RAG-Prozess

Erstellt: 2025-12-24 | Aktualisiert: 2025-12-31

Status: Aktualisiert 2025-12-31

Übersicht

Der RAG-Prozess (Retrieval Augmented Generation) bildet das Herzstück der KI-gestützten Wissensgenerierung. Die Architektur trennt strikt zwischen Offline-Pipeline (Wissensaufbau) und Online-Pipeline (Wissenskonsum).

Grundprinzip

• Die Offline-Pipeline erzeugt und strukturiert Wissen (Python-Skripte)
• Die Online-Pipeline konsumiert Wissen, ohne es zu verändern (PHP-Services)
• Zwischen beiden Pipelines besteht keine Rückkopplung zur Laufzeit

Offline-Pipeline: Tatsächlicher Ablauf (IST)

Pfad: /var/www/scripts/pipeline/
Orchestrator: pipeline.py
Step-Module: step_extract.py, step_load.py, step_transform.py, step_embed.py, step_semantic.py

Phase 0: Detection

Phase 1: Extraction (pro Dokument)

Phase 2: Load

Phase 3: Vision (nur PDF)

Phase 4: Chunking

Phase 5: Enrichment (nur PDF)

Phase 6: Embedding (Layer 3 - Dokument wird suchbar)

Nach Phase 6 ist das Dokument im Chat suchbar!

Phase 7: Semantic Analysis (Layer 4 - optional/async)

Phase 8: Finalization

Datenfluss-Diagramm

┌─────────────────────────────────────────────────────────────────┐
│ Phase 0: DETECTION                                               │
│ detect.py: scan_directory() → queue_files()                     │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│ Phase 1: EXTRACTION                                              │
│ step_extract.py → extract.py                                    │
│   • Text aus PDF/DOCX/PPTX/MD/TXT                               │
│   • SHA256 Hash                                                 │
│   • Rotationserkennung (orientation.py)                         │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│ Phase 2: LOAD                                                    │
│ step_load.py                                                    │
│   • documents Eintrag erstellen                                 │
│   • document_pages speichern                                    │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼ (nur PDF)
┌─────────────────────────────────────────────────────────────────┐
│ Phase 3: VISION                                                  │
│ step_transform.py → vision.py                                   │
│   • Seiten als Bilder rendern                                   │
│   • llama3.2-vision:11b Analyse                                 │
│   • Überschriften, Bilder, Tabellen, Layout erkennen            │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│ Phase 4: CHUNKING                                                │
│ step_transform.py → chunk.py                                    │
│   • Semantisches Chunking nach Struktur                         │
│   • Überlappung: 10%                                            │
│   • Min: 100, Max: 2000 Zeichen                                 │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼ (nur PDF)
┌─────────────────────────────────────────────────────────────────┐
│ Phase 5: ENRICHMENT                                              │
│ step_transform.py → enrich.py                                   │
│   • Vision-Kontext zu Chunks hinzufügen                         │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│ Phase 6: EMBEDDING (Layer 3)                                     │
│ step_embed.py → embed.py                                        │
│   • mxbai-embed-large (1024 dim)                                │
│   • Qdrant: Collection "documents"                              │
│   • Status: "embedded" ✓ SUCHBAR                                │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│ Phase 7: SEMANTIC (Layer 4) - Optional                           │
│ step_semantic.py → analyzers/*                                  │
│   • Entity-Extraktion (Anthropic oder Ollama)                   │
│   • Relationen                                                  │
│   • Taxonomie-Klassifikation                                    │
│   • Ontologie-Zuordnung                                         │
│   • Chunk-Semantik (summary, keywords, sentiment)               │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│ Phase 8: FINALIZATION                                            │
│   • Status: "done" ✓                                            │
└─────────────────────────────────────────────────────────────────┘

Konfiguration

Quelle: config.py

Datei-Referenzen

Pipeline-Kern

/var/www/scripts/pipeline/
├── pipeline.py           # Orchestrator
├── config.py             # Konfiguration
├── db.py                 # Datenbank-Wrapper
├── detect.py             # Datei-Erkennung
├── extract.py            # Text-Extraktion
├── orientation.py        # Rotationserkennung (NEU!)
├── chunk.py              # Chunking
├── embed.py              # Embedding
├── vision.py             # Vision-Analyse
├── enrich.py             # Vision-Enrichment

Step-Module

├── step_extract.py       # Extraktions-Step
├── step_load.py          # Lade-Step
├── step_transform.py     # Transform-Step (Vision, Chunk, Enrich)
├── step_embed.py         # Embedding-Step
└── step_semantic.py      # Semantik-Step

Analyzer-Module

├── analyzers/
│   ├── document_analyzer.py     # Haupt-Orchestrator
│   ├── entity_extractor.py      # Entitäten
│   ├── entity_normalizer.py     # Normalisierung
│   ├── relation_extractor.py    # Relationen
│   ├── taxonomy_classifier.py   # Taxonomie
│   ├── ontology_classifier.py   # Ontologie
│   ├── semantic_analyzer.py     # Chunk-Semantik
│   └── statement_analyzer.py    # SPO-Aussagen

CLI-Befehle

# Vollständiger Durchlauf
python pipeline.py all

# Mit Tracking (für Web-UI)
python pipeline.py all --pipeline-id=4 --run-id=6

# Einzelne Phasen
python pipeline.py scan      # Nur scannen
python pipeline.py process   # Queue abarbeiten
python pipeline.py embed     # Ausstehende Embeddings
python pipeline.py semantic 123  # Semantik für Dokument 123
python pipeline.py status    # Status anzeigen

# Einzeldatei
python pipeline.py file /path/to/document.pdf

Datenbank-Tabellen

Content-Datenbank (ki_content)

Dev-Datenbank (ki_dev)

Referenz

• Task #442: PDF-Rotationserkennung implementiert
• Doc #143: PDF-Rotationserkennung Details

Verwandte Themen

• Prozesse - Übersicht aller Prozesse
• Scheduling - Cron-Jobs für Pipeline

Quality Gates

Erstellt: 2025-12-23 | Aktualisiert: 2025-12-28

Automatische Code-Qualitätsprüfung via Claude Code Hooks mit Task-Integration.

/var/www/tools/ki-protokoll/claude-hook/
├── hook_dispatcher.py      ← Stabiler Einstiegspunkt
├── .env                    ← DB-Credentials
└── quality/
    ├── pre_rules.py            ← BLOCK-Regeln Dispatcher
    ├── post_rules.py           ← WARN-Regeln Dispatcher
    ├── pre_rules_guard.py      ← P1.x SRP, P4.x OOP
    ├── pre_rules_mvc.py        ← P2.x MVC
    ├── pre_rules_validation.py ← P3.x PSR
    ├── pre_rules_layers.py     ← P6.x SOLID
    ├── pre_rules_htmx.py       ← HTMX-C1-C5
    ├── pre_rules_constants.py  ← P7.x Magic Numbers
    ├── pre_rules_deterministic.py ← P8.x Deterministic
    ├── pre_rules_tests.py      ← P14.x Test Isolation
    ├── pre_rules_python.py     ← PP1.x Python
    ├── rules_quality.py        ← W1.x-W6.x Quality
    ├── rules_security.py       ← Security Warnings
    ├── rules_style.py          ← Style Warnings
    ├── rules_constants.py      ← W7.x Magic Numbers
    ├── rules_failfast.py       ← W8.x Fail Fast
    ├── rules_failsafe.py       ← W9.x Fail Safe (Phase 3)
    ├── rules_tradeoffs.py      ← W10.x Trade-offs (Phase 3)
    ├── rules_leastsurprise.py  ← W15.x Least Surprise
    ├── rules_testisolation.py  ← W14.x Test Isolation
    ├── phpmetrics_check.py     ← W13.x Cohesion (Phase 3)
    └── task_creator.py         ← Violations → Tasks

# Installation
/opt/php-tools/vendor/bin/phpmetrics

# Analyse ausführen
/opt/php-tools/vendor/bin/phpmetrics --report-json=/tmp/phpmetrics.json /var/www/dev.campus.systemische-tools.de/src

# Cohesion-Check (Hook-Integration)
python3 /var/www/tools/ki-protokoll/claude-hook/quality/phpmetrics_check.py

Chat UI Redesign: 3-Spalten-Layout

Erstellt: 2025-12-27 | Aktualisiert: 2025-12-27

Task #424 - KORRIGIERTER Implementierungsplan
Version: 2.0 (nach kritischem Review)
Datum: 2025-12-27

Architektur-Entscheidungen

Wiederverwendung bestehender Infrastruktur

Korrigierte Prinzipien

Dateien und Änderungen

1. NEUER Controller: src/Controller/Api/ConfigController.php

<?php

declare(strict_types=1);

namespace Controller\Api;

// @responsibility: JSON-API für Config-Operationen (Chat-Integration)

use Framework\Controller;
use UseCases\Config\ManageConfigUseCase;

class ConfigController extends Controller
{
    public function __construct(
        private ManageConfigUseCase $configUseCase
    ) {
    }

    /**
     * GET /api/v1/config/{id}
     * Lädt Config mit Content für Anzeige
     */
    public function show(string $id): void
    {
        $config = $this->configUseCase->getById((int) $id);
        
        if ($config === null) {
            $this->json(['error' => 'Config nicht gefunden'], 404);
            return;
        }

        $this->json([
            'id' => $config->id,
            'name' => $config->name,
            'type' => $config->type,
            'content' => $config->content,
            'version' => $config->version,
        ]);
    }

    /**
     * POST /api/v1/config/{id}
     * Speichert neue Version (auto-increment)
     */
    public function update(string $id): void
    {
        $this->requireCsrf();

        $config = $this->configUseCase->getById((int) $id);
        if ($config === null) {
            $this->json(['error' => 'Config nicht gefunden'], 404);
            return;
        }

        $newContent = $_POST['content'] ?? '';
        
        // Auto-increment Version
        $currentVersion = $config->version ?? '1.0';
        $versionParts = explode('.', $currentVersion);
        $newVersion = $versionParts[0] . '.' . ((int) ($versionParts[1] ?? 0) + 1);

        $result = $this->configUseCase->update(
            id: (int) $id,
            name: $config->name,
            slug: $config->slug,
            description: $config->description,
            content: $newContent,
            newVersion: $newVersion,
            changeDescription: 'Chat UI Update',
            status: $config->status,
            parentId: $config->parentId
        );

        if (!$result->success) {
            $this->json(['error' => $result->message], 400);
            return;
        }

        $this->json([
            'success' => true,
            'version' => $newVersion,
            'message' => "Version {$newVersion} gespeichert",
        ]);
    }
}

Zeilen: ~70 | Dependencies: 1 (ManageConfigUseCase) | SRP: Nur Config-API

2. Routes: routes/api.php

Hinzufügen:

// Config API (für Chat-Integration)
use Controller\Api\ConfigController;

$router->get('/api/v1/config/{id}', [ConfigController::class, 'show']);
$router->post('/api/v1/config/{id}', [ConfigController::class, 'update']);

3. Interface erweitern: src/Domain/Repository/ContentConfigRepositoryInterface.php

interface ContentConfigRepositoryInterface
{
    // ... bestehende Methoden ...

    /**
     * Save version to history before update.
     */
    public function saveHistory(
        int $configId,
        string $content,
        string $version,
        string $changedBy,
        string $changeDescription
    ): void;

    /**
     * Get version history for a config.
     *
     * @return array<int, array<string, mixed>>
     */
    public function getHistory(int $configId, int $limit = 10): array;
}

4. View: src/View/chat/index.php

ENTFERNEN aus .chat-settings--advanced:

• select[name="author_profile_id"]
• select[name="system_prompt_id"]
• select[name="structure_id"]

NEU einfügen nach <aside class="chat-sidebar">...</aside>:

<!-- Config Panel -->
<aside class="chat-config" id="configPanel">
    <div class="chat-config__header">
        <span>Konfiguration</span>
    </div>

    <!-- System Prompt -->
    <div class="chat-config__section" data-config-type="system_prompt">
        <label for="configSystemPrompt" class="chat-config__label">System Prompt</label>
        <div class="chat-config__row">
            <select name="system_prompt_id" id="configSystemPrompt" class="chat-config__select">
                <?php foreach ($systemPrompts ?? [] as $prompt): ?>
                <option value="<?= $prompt['id'] ?>"
                        data-content="<?= htmlspecialchars($prompt['content'] ?? '{}') ?>">
                    <?= htmlspecialchars($prompt['name']) ?>
                </option>
                <?php endforeach; ?>
            </select>
            <button type="button" 
                    class="chat-config__toggle"
                    aria-expanded="false"
                    aria-controls="systemPromptEdit"
                    aria-label="System Prompt bearbeiten">
                <span aria-hidden="true">▸</span>
            </button>
        </div>
        <div class="chat-config__edit" id="systemPromptEdit" hidden>
            <textarea id="systemPromptText" class="chat-config__textarea" rows="8"></textarea>
            <button type="button" class="chat-config__save" data-config-id="">
                Speichern
            </button>
        </div>
    </div>
    <!-- ... weitere Sections (Struktur, Autor-Profil) ... -->
</aside>

5. CSS: public/css/chat-redesign.css

/* CONFIG PANEL (3-Spalten-Layout) */

.chat-config {
    width: 300px;
    background: var(--chat-sidebar-bg);
    border-right: 1px solid var(--chat-border);
    display: flex;
    flex-direction: column;
    flex-shrink: 0;
    overflow-y: auto;
}

.chat-config__header {
    padding: 16px;
    font-size: 12px;
    font-weight: 600;
    text-transform: uppercase;
    letter-spacing: 0.5px;
    color: var(--chat-text-secondary);
    border-bottom: 1px solid var(--chat-border);
}

.chat-config__section { padding: 12px 16px; }
.chat-config__label { display: block; font-size: 11px; margin-bottom: 6px; }
.chat-config__row { display: flex; gap: 8px; align-items: center; }
.chat-config__select { flex: 1; padding: 8px 10px; border-radius: 6px; }
.chat-config__toggle { padding: 6px 10px; border-radius: 6px; cursor: pointer; }
.chat-config__textarea { width: 100%; font-family: monospace; font-size: 12px; }
.chat-config__save { width: 100%; padding: 8px; background: var(--chat-accent); }

@media (max-width: 1024px) {
    .chat-config { display: none; }
}

Zusammenfassung der Änderungen

Total neue Zeilen: ~310 | Geänderte Dateien: 5 | Neue Dateien: 1

Checkliste vor Implementation

• ManageConfigUseCase im DI-Container registriert?
• PHPStan Level 5+ prüfen
• WCAG: Alle ARIA-Attribute korrekt?
• Contract #5 (layered-architecture): Kein SQL im Controller?
• Contract #4 (js-browser): Keine == Vergleiche?
• Contract #2 (css): Focus-Styles vorhanden?

Nicht implementiert (YAGNI)

• Version-History-Browser → Nur Save = neue Version
• History-Wiederherstellung → Via /config/{id} Web-UI
• Komplexe Version-Auswahl → Automatisches Inkrement

Verwandte Themen

• HTMX - Interaktivitäts-Patterns
• Architektur - Clean Architecture

db.py Refactoring Plan

Status

• Task ID: 505
• Erstellt: 2025-12-28
• Status: Plan erstellt

Ausgangslage

Datei: /var/www/scripts/pipeline/db.py
Zeilen: 834
Klassen: 2 (Database: 707 Zeilen, PipelineProgress: 109 Zeilen)
Problem: 14 verschiedene Verantwortlichkeiten in der Database-Klasse (SRP-Verletzung)

Ziele

1. Jedes Modul unter 500 Zeilen (Code Hygiene Limit)
2. Strikte Einhaltung von SRP (Single Responsibility Principle)
3. 100% Rückwärtskompatibilität - alle bestehenden Imports funktionieren weiterhin
4. DRY, KISS, SOLID Prinzipien

Architektur: Mixin-Pattern

Python Mixins ermöglichen das Aufteilen einer Klasse in logische Einheiten bei gleichzeitiger Beibehaltung der Abwärtskompatibilität.

db_core.py          ─┐
db_documents.py      │
db_queue.py          ├── Mixins ──► db.py (Database erbt von allen)
db_logging.py        │
db_semantic.py       │
db_prompts.py       ─┘

Modul-Aufteilung

1. db_core.py (~100 Zeilen)

Verantwortung: Connection Management

class DatabaseCore:
    def __init__(self): ...
    def connect(self): ...
    def disconnect(self): ...
    def execute(self, query, params=None): ...
    def commit(self): ...

2. db_documents.py (~150 Zeilen)

Verantwortung: Document, Page, Chunk CRUD

class DocumentsMixin:
    # Documents
    def document_exists(self, file_path): ...
    def document_is_done(self, file_path): ...
    def insert_document(self, file_path, title, file_type, file_size, file_hash): ...
    def update_document_status(self, doc_id, status, error_message=None): ...
    
    # Pages
    def insert_page(self, doc_id, page_number, text_content, token_count=None): ...
    def get_page_id(self, doc_id, page_number): ...
    
    # Chunks
    def insert_chunk(self, doc_id, chunk_index, content, heading_path, ...): ...
    def get_chunks_for_embedding(self, limit=DEFAULT_LIMIT): ...
    def update_chunk_qdrant_id(self, chunk_id, qdrant_id): ...

3. db_queue.py (~60 Zeilen)

Verantwortung: Pipeline Queue Operations

class QueueMixin:
    def add_to_queue(self, file_path, action="process"): ...
    def get_pending_queue_items(self, limit=10): ...
    def update_queue_status(self, queue_id, status, error_message=None): ...

4. db_logging.py (~180 Zeilen)

Verantwortung: Alle Logging-Operationen

class LoggingMixin:
    def log(self, level, message, context=None): ...
    def log_to_protokoll(self, client_name, request, response=None, ...): ...
    def log_provenance(self, artifact_type, artifact_id, source_type, ...): ...

5. db_semantic.py (~250 Zeilen)

Verantwortung: Entity Types, Stopwords, Taxonomy, Synonyms

class SemanticMixin:
    # Entity Types
    def get_entity_types(self, active_only=True): ...
    def get_entity_type_codes(self): ...
    def build_entity_prompt_categories(self): ...
    
    # Stopwords
    def get_stopwords(self, active_only=True): ...
    def is_stopword(self, word): ...
    def _normalize_stopword(self, word): ...
    
    # Synonyms (internal, nicht extern genutzt)
    def find_entity_by_synonym(self, synonym): ...
    def add_synonym(self, entity_id, synonym, ...): ...
    
    # Chunk Taxonomy
    def add_chunk_taxonomy(self, chunk_id, term_id, ...): ...
    def get_chunk_taxonomies(self, chunk_id): ...
    
    # Entity Taxonomy
    def add_entity_taxonomy(self, entity_id, term_id, ...): ...
    def get_entity_taxonomies(self, entity_id): ...
    def get_taxonomy_terms(self): ...

6. db_prompts.py (~70 Zeilen)

Verantwortung: Prompt-Verwaltung

class PromptsMixin:
    def get_prompt(self, name, version=None): ...
    def get_prompt_by_use_case(self, use_case, version=None): ...

7. db.py (~100 Zeilen - Kompositions-Layer)

Verantwortung: Rückwärtskompatibilität

from db_core import DatabaseCore
from db_documents import DocumentsMixin
from db_queue import QueueMixin
from db_logging import LoggingMixin
from db_semantic import SemanticMixin
from db_prompts import PromptsMixin

class Database(
    DatabaseCore,
    DocumentsMixin,
    QueueMixin,
    LoggingMixin,
    SemanticMixin,
    PromptsMixin
):
    '''Vollständige Database-Klasse mit allen Operationen.'''
    pass

class PipelineProgress:
    # ... (unverändert, eigene Klasse)

# Globale Instanz für Rückwärtskompatibilität
db = Database()

Zeilen-Verteilung

Rückwärtskompatibilität

Alle existierenden Imports funktionieren weiterhin:

# Diese Imports bleiben gültig:
from db import db
from db import Database
from db import PipelineProgress
from db import db, PipelineProgress

Implementierungs-Reihenfolge

1. db_core.py erstellen (Basis für alle anderen)
2. db_logging.py erstellen (wird von vielen benötigt)
3. db_prompts.py erstellen (eigenständig)
4. db_documents.py erstellen (eigenständig)
5. db_queue.py erstellen (eigenständig)
6. db_semantic.py erstellen (eigenständig)
7. db.py umschreiben (Komposition)
8. Tests durchführen (alle Pipeline-Scripts testen)

Risiken & Mitigationen

Cleanup-Möglichkeiten

• Entfernen: find_entity_by_synonym, add_synonym (nicht extern genutzt)
• Vereinfachen: get_page_id könnte in insert_page integriert werden
• Optimieren: Redundante Cursor-Handling kann vereinheitlicht werden

Erstellt: 2025-12-22 | Aktualisiert: 2025-12-27

Refactoring-Session 22.12.2025 - Abgeschlossen

Zusammenfassung

Umfangreiche Architektur-Verbesserungen basierend auf externer Supervision. Alle 9 Phasen erfolgreich implementiert.

Abgeschlossene Phasen

Neue Komponenten (Phase 3-4)

Formatter (src/Infrastructure/Formatter/)

• ApiResponseFormatter.php - Einheitliche API-Responses (success, error, list, notFound)
• DateFormatter.php - Konsistente Datumsformatierung (short, dateTime, relative)

State Machines (src/Domain/ValueObject/)

• TaskStatus.php - 5 States mit Transition-Validierung
• ContentOrderStatus.php - 7 States (draft bis published)
• PipelineRunStatus.php - 5 States fuer Pipeline-Runs

Logging (src/Infrastructure/Logging/)

• AuditLogger.php - Zentrale Logging-Klasse mit Correlation-ID

Middleware (src/Framework/Middleware/)

• CorrelationMiddleware.php - Request-Tracing via X-Correlation-ID

Exception (src/Domain/Exception/)

• InvalidStateTransitionException.php - Fuer State Machine Violations

Datenbank

• audit_log Tabelle in ki_dev mit Indizes fuer correlation_id, event, created_at

Validierung Phase 6

Contract-Validierung

• db-access-security-protocol: PASSED (0 Violations)
• code-quality-standards: 4 major (vorbestehende God-Classes)

Controller LOC-Pruefung

15 von 17 Controllern unter 300 LOC. ContentController (397) und ContentPipelineController (631) noch zu gross.

PDO-Pruefung

Keine direkten PDO-Zugriffe in Controllern - OK

PHP-Quality-Check

• PHPStan: OK (keine Fehler)
• PHP-CS-Fixer: OK (0 Fixes)
• Composer Audit: Keine Schwachstellen
• Semgrep: Keine Schwachstellen

Bekannte Probleme (vorbestehend)

• 4 God-Classes ueber 500 LOC (SemanticExplorerRepository, ContentPipelineController, ChunkSyncService, SystemExplorerRepository)
• Unit-Tests benoetigen Autoload-Konfiguration in composer.json

Architektur nach Refactoring

Request
   |
   v
CorrelationMiddleware (X-Correlation-ID)
   |
   v
Router -> Controller
              |
              v
         Application Service
              |
              +-> Repository (DB via DatabaseFactory)
              |
              +-> Domain ValueObjects (State Machines)
              |
              +-> AuditLogger (Logging)
              |
              v
         Formatter (API/Date)
              |
              v
         Response (View/JSON/Redirect)