{
"event": "PreToolUse",
"tool_name": "mcp__mcp-tasks__tasks_update",
"tool_input": {
"id": 516,
"description": "# Hardcodierte Model-Werte in Pipeline entfernen\n\n## Status: GENEHMIGT MIT AUFLAGEN (Supervision 2025-12-28)\n\n---\n\n## 1. PROBLEM-BESCHREIBUNG\n\nDie Python-Pipeline (`\/var\/www\/scripts\/pipeline\/`) hat hardcodierte LLM-Modellnamen, die die zentrale Konfiguration in `pipeline_steps.config` (Datenbank) umgehen.\n\n### Symptom\n- Änderung des Modells in der GUI\/DB hat keine Wirkung\n- Code verwendet Fallback-Werte statt DB-Konfiguration\n\n### Root Cause\nHardcodierte Fallback-Werte in Python-Dateien:\n```python\n# Beispiel aus step_semantic_extended.py:126\nmodel = config.get(\"model\", \"mistral\") # ← Fallback \"mistral\"\n```\n\n---\n\n## 2. IST-ZUSTAND: Betroffene Dateien\n\n### 2.1 KRITISCH: Pipeline-Kernlogik\n\n| Datei | Zeile | Code | Auswirkung |\n|-------|-------|------|------------|\n| `pipeline.py` | 242 | `{\"model\": \"mistral\"}` | Text-Semantik ignoriert DB |\n| `pipeline.py` | 286 | `{\"model\": \"mistral\"}` | Knowledge-Semantik ignoriert DB |\n| `pipeline.py` | 557 | `model = \"mistral:latest\"` | Entity-Enrichment ignoriert DB |\n| `step_semantic_extended.py` | 126 | `config.get(\"model\", \"mistral\")` | Fallback statt Exception |\n| `step_semantic_extended.py` | 286 | `config.get(\"model\", \"mistral\")` | Fallback statt Exception |\n| `step_entity_enrich.py` | 20 | `DEFAULT_MODEL = \"mistral:latest\"` | Konstante statt DB |\n\n### 2.2 MITTEL: Standalone-Scripts\n\n| Datei | Zeile | Klassifikation | Entscheidung |\n|-------|-------|----------------|--------------|\n| `generate_entity_descriptions.py` | 30 | **PRODUKTIV** | → DB anbinden |\n| `generate_semantics.py` | 14 | **PRODUKTIV** | → DB anbinden |\n| `backfill_*.py` | - | **PRODUKTIV** | ✅ Bereits korrekt |\n\n---\n\n## 3. SUPERVISION-AUFLAGEN (integriert)\n\n### 3.1 ✅ Pipeline-ID Hardcoding entfernen\n\n**Problem**: `pipeline_id: int = 5` ist eine neue Hardcodierung.\n\n**Lösung**: Pipeline-ID aus Aufrufkontext oder Environment:\n```python\nimport os\n\ndef get_step_model(step_type: str, pipeline_id: int = None) -> str:\n if pipeline_id is None:\n pipeline_id = int(os.environ.get(\"PIPELINE_ID\", \"5\"))\n # ...\n```\n\n### 3.2 ✅ Mehrfachzeilen-Fall absichern\n\n**Problem**: Mehrere Zeilen für (pipeline_id, step_type) = Datenfehler.\n\n**Lösung A**: UNIQUE Constraint prüfen\/erstellen:\n```sql\n-- Prüfen ob existiert:\nSHOW INDEX FROM pipeline_steps WHERE Column_name = 'step_type';\n\n-- Falls nicht: Constraint erstellen\nALTER TABLE pipeline_steps \nADD UNIQUE INDEX uq_pipeline_step (pipeline_id, step_type);\n```\n\n**Lösung B**: Im Code absichern:\n```python\nrows = cursor.fetchall()\nif len(rows) > 1:\n raise PipelineConfigError(\n f\"Datenfehler: Mehrere Einträge für pipeline_id={pipeline_id}, \"\n f\"step_type='{step_type}'. UNIQUE Constraint fehlt!\"\n )\nif len(rows) == 0:\n raise PipelineConfigError(...)\n```\n\n### 3.3 ✅ Tests erweitern\n\n**Zusätzliche Tests (Supervision-Anforderung):**\n\n| Test | Zweck |\n|------|-------|\n| `test_no_hardcoded_models.py` | Scannt pipeline.py auf Literal-Modelle |\n| `test_step_config_complete.py` | Prüft ob jeder Step-Call `model` in config hat |\n| `test_step_type_mapping.py` | Prüft ob step_type Strings korrekt sind (Tippfehler = Ausfall) |\n\n### 3.4 ✅ Standalone-Scripts klassifizieren\n\n| Script | Nutzung | Entscheidung |\n|--------|---------|--------------|\n| `generate_entity_descriptions.py` | Produktiv (Cron) | DB anbinden |\n| `generate_semantics.py` | Produktiv | DB anbinden |\n| `backfill_*.py` | Produktiv | ✅ Bereits korrekt |\n| `quality_test.py` | Dev-only | Explizit ausnehmen |\n\n---\n\n## 4. KORRIGIERTER LÖSUNGSANSATZ\n\n### 4.1 Zentrales Modul: `pipeline_config.py`\n\n```python\n#!\/usr\/bin\/env python3\n\"\"\"\nZentrale Pipeline-Konfiguration aus DB.\n\nSingle Source of Truth: ki_content.pipeline_steps.config\n\"\"\"\n\nimport os\nfrom db import db\n\n\nclass PipelineConfigError(Exception):\n \"\"\"Fehler wenn Config fehlt - KEIN stilles Fallback!\"\"\"\n pass\n\n\ndef get_step_model(step_type: str, pipeline_id: int = None) -> str:\n \"\"\"\n Liest Model aus pipeline_steps.config.\n \n Args:\n step_type: Step-Typ (z.B. 'text_semantic_analyze')\n pipeline_id: Pipeline-ID (Default aus ENV oder 5)\n \n Returns:\n Model-Name als String\n \n Raises:\n PipelineConfigError: Bei fehlendem\/ungültigem Config\n \n WICHTIG:\n - Wirft Exception wenn nicht konfiguriert!\n - Kein stilles Fallback auf hardcodierte Werte.\n - Mehrfachzeilen sind ein Datenfehler.\n \"\"\"\n if pipeline_id is None:\n pipeline_id = int(os.environ.get(\"PIPELINE_ID\", \"5\"))\n \n cursor = db.execute(\n \"\"\"SELECT JSON_UNQUOTE(JSON_EXTRACT(config, '$.model')) as model\n FROM pipeline_steps\n WHERE pipeline_id = %s AND step_type = %s AND enabled = 1\"\"\",\n (pipeline_id, step_type)\n )\n rows = cursor.fetchall()\n cursor.close()\n \n # Mehrfachzeilen = Datenfehler\n if len(rows) > 1:\n raise PipelineConfigError(\n f\"DATENFEHLER: {len(rows)} Einträge für \"\n f\"pipeline_id={pipeline_id}, step_type='{step_type}'. \"\n f\"Erwarte genau 1. UNIQUE Constraint prüfen!\"\n )\n \n # Keine Zeile = Config fehlt\n if len(rows) == 0:\n raise PipelineConfigError(\n f\"Kein Model konfiguriert für step_type='{step_type}' \"\n f\"in pipeline_id={pipeline_id}. \"\n f\"Bitte in pipeline_steps.config setzen!\"\n )\n \n model = rows[0].get(\"model\")\n \n # Leerer String = ungültig\n if not model or model.strip() == \"\":\n raise PipelineConfigError(\n f\"Model ist leer für step_type='{step_type}' \"\n f\"in pipeline_id={pipeline_id}.\"\n )\n \n return model.strip()\n```\n\n### 4.2 Änderungen pro Datei\n\n**pipeline.py**\n```python\n# Zeile 1: Import hinzufügen\nfrom pipeline_config import get_step_model\n\n# Zeile 242: ALT → NEU\n# ALT: analyzed_chunks = text_semantic_analyze.execute(chunk_data, {\"model\": \"mistral\"})\nmodel = get_step_model(\"text_semantic_analyze\")\nanalyzed_chunks = text_semantic_analyze.execute(chunk_data, {\"model\": model})\n\n# Zeile 286: ALT → NEU\n# ALT: analyzed_entities = knowledge_semantic_analyze.execute(entity_list, {\"model\": \"mistral\"})\nmodel = get_step_model(\"knowledge_semantic_analyze\")\nanalyzed_entities = knowledge_semantic_analyze.execute(entity_list, {\"model\": model})\n\n# Zeile 557: ALT → NEU\n# ALT: model = \"mistral:latest\"\nmodel = get_step_model(\"entity_enrich\")\n```\n\n**step_semantic_extended.py**\n```python\n# Zeile 126 und 286: ALT → NEU\n# ALT: model = config.get(\"model\", \"mistral\")\nmodel = config.get(\"model\")\nif not model:\n raise ValueError(\n f\"Model muss in config übergeben werden! \"\n f\"Step: {self.__class__.__name__}\"\n )\n```\n\n---\n\n## 5. DB-VORBEREITUNG\n\n### 5.1 UNIQUE Constraint prüfen\n\n```sql\n-- Status prüfen\nSELECT COUNT(*) as duplicates\nFROM pipeline_steps\nGROUP BY pipeline_id, step_type\nHAVING COUNT(*) > 1;\n\n-- Constraint erstellen (falls keine Duplikate)\nALTER TABLE pipeline_steps \nADD UNIQUE INDEX uq_pipeline_step (pipeline_id, step_type);\n```\n\n### 5.2 Step-Type für entity_enrich hinzufügen\n\n```sql\n-- Prüfen ob existiert\nSELECT * FROM pipeline_steps \nWHERE pipeline_id = 5 AND step_type = 'enrich';\n\n-- Falls nicht: Hinzufügen\nINSERT INTO pipeline_steps (pipeline_id, step_type, config, sort_order, enabled)\nVALUES (5, 'enrich', '{\"model\": \"mistral\"}', 50, 1);\n```\n\n---\n\n## 6. ERWEITERTER TEST-PLAN\n\n### 6.1 Neue Tests (Supervision-Anforderung)\n\n```python\n# test_no_hardcoded_models.py\ndef test_pipeline_no_literal_models():\n \"\"\"Scannt pipeline.py auf hardcodierte Model-Strings.\"\"\"\n import re\n with open(\"pipeline.py\") as f:\n content = f.read()\n \n # Suche nach {\"model\": \"...\"} oder model = \"...\"\n pattern = r'[\"\\']model[\"\\']\\s*:\\s*[\"\\'][^\"\\']+[\"\\']'\n matches = re.findall(pattern, content)\n \n assert len(matches) == 0, f\"Hardcodierte Models gefunden: {matches}\"\n\n\n# test_step_type_mapping.py\ndef test_step_types_exist_in_db():\n \"\"\"Prüft ob alle verwendeten step_types in DB existieren.\"\"\"\n from pipeline_config import get_step_model\n from db import db\n \n db.connect()\n \n step_types = [\n \"text_semantic_analyze\",\n \"knowledge_semantic_analyze\",\n \"entity_extract\",\n \"enrich\",\n ]\n \n for st in step_types:\n try:\n model = get_step_model(st)\n assert model, f\"Model leer für {st}\"\n except Exception as e:\n pytest.fail(f\"step_type '{st}' nicht konfiguriert: {e}\")\n \n db.disconnect()\n```\n\n### 6.2 Integration-Tests\n\n- [ ] Pipeline-Run mit DB-Config → verwendet DB-Model\n- [ ] DB-Update auf anderes Model → nächster Run verwendet neues Model\n- [ ] Fehlender Config → Exit Code != 0, Fehlermeldung enthält step_type\n\n### 6.3 Manuelle Tests\n\n- [ ] GUI: Model ändern → Pipeline verwendet neues Model\n- [ ] CLI: `python pipeline.py process 1` ohne Fehler\n- [ ] ENV: `PIPELINE_ID=5 python pipeline.py process 1` funktioniert\n\n---\n\n## 7. RISIKO-BEWERTUNG (aktualisiert)\n\n| Risiko | Bewertung | Mitigation |\n|--------|-----------|------------|\n| Pipeline-Ausfall bei fehlendem Config | GEWOLLT | Exception macht Konfigfehler sichtbar |\n| Mehrfachzeilen in DB | MITTEL | UNIQUE Constraint + Code-Prüfung |\n| Tippfehler in step_type | HOCH | Test `test_step_type_mapping.py` |\n| Breaking Change CLI-Scripts | MITTEL | Produktive Scripts anbinden, Dev-only ausnehmen |\n\n---\n\n## 8. IMPLEMENTIERUNGS-REIHENFOLGE\n\n1. **DB vorbereiten**\n - UNIQUE Constraint prüfen\/erstellen\n - Fehlende step_types hinzufügen (z.B. `enrich`)\n\n2. **pipeline_config.py erstellen**\n - Mit allen Supervision-Korrekturen\n\n3. **Tests erstellen**\n - `test_no_hardcoded_models.py`\n - `test_step_type_mapping.py`\n\n4. **Code ändern**\n - pipeline.py\n - step_semantic_extended.py\n - step_entity_enrich.py\n - generate_entity_descriptions.py\n - generate_semantics.py\n\n5. **Integration-Test**\n - Kompletter Pipeline-Durchlauf\n\n---\n\n## 9. SUPERVISION-ENTSCHEIDUNGEN (dokumentiert)\n\n| Frage | Entscheidung | Begründung |\n|-------|--------------|-----... [TRUNCATED-fc80a2260189ab66]"
}
}
{
"tool_response": [
{
"type": "text",
"text": "{\n \"success\": true,\n \"task\": {\n \"id\": 516,\n \"uuid\": \"e293c47c-ebe7-48d0-97cc-1890e4805a60\",\n \"title\": \"Hardcodierte Model-Werte in Pipeline entfernen\",\n \"description\": \"# Hardcodierte Model-Werte in Pipeline entfernen\\n\\n## Status: GENEHMIGT MIT AUFLAGEN (Supervision 2025-12-28)\\n\\n---\\n\\n## 1. PROBLEM-BESCHREIBUNG\\n\\nDie Python-Pipeline (`\/var\/www\/scripts\/pipeline\/`) hat hardcodierte LLM-Modellnamen, die die zentrale Konfiguration in `pipeline_steps.config` (Datenbank) umgehen.\\n\\n### Symptom\\n- Änderung des Modells in der GUI\/DB hat keine Wirkung\\n- Code verwendet Fallback-Werte statt DB-Konfiguration\\n\\n### Root Cause\\nHardcodierte Fallback-Werte in Python-Dateien:\\n```python\\n# Beispiel aus step_semantic_extended.py:126\\nmodel = config.get(\\\"model\\\", \\\"mistral\\\") # ← Fallback \\\"mistral\\\"\\n```\\n\\n---\\n\\n## 2. IST-ZUSTAND: Betroffene Dateien\\n\\n### 2.1 KRITISCH: Pipeline-Kernlogik\\n\\n| Datei | Zeile | Code | Auswirkung |\\n|-------|-------|------|------------|\\n| `pipeline.py` | 242 | `{\\\"model\\\": \\\"mistral\\\"}` | Text-Semantik ignoriert DB |\\n| `pipeline.py` | 286 | `{\\\"model\\\": \\\"mistral\\\"}` | Knowledge-Semantik ignoriert DB |\\n| `pipeline.py` | 557 | `model = \\\"mistral:latest\\\"` | Entity-Enrichment ignoriert DB |\\n| `step_semantic_extended.py` | 126 | `config.get(\\\"model\\\", \\\"mistral\\\")` | Fallback statt Exception |\\n| `step_semantic_extended.py` | 286 | `config.get(\\\"model\\\", \\\"mistral\\\")` | Fallback statt Exception |\\n| `step_entity_enrich.py` | 20 | `DEFAULT_MODEL = \\\"mistral:latest\\\"` | Konstante statt DB |\\n\\n### 2.2 MITTEL: Standalone-Scripts\\n\\n| Datei | Zeile | Klassifikation | Entscheidung |\\n|-------|-------|----------------|--------------|\\n| `generate_entity_descriptions.py` | 30 | **PRODUKTIV** | → DB anbinden |\\n| `generate_semantics.py` | 14 | **PRODUKTIV** | → DB anbinden |\\n| `backfill_*.py` | - | **PRODUKTIV** | ✅ Bereits korrekt |\\n\\n---\\n\\n## 3. SUPERVISION-AUFLAGEN (integriert)\\n\\n### 3.1 ✅ Pipeline-ID Hardcoding entfernen\\n\\n**Problem**: `pipeline_id: int = 5` ist eine neue Hardcodierung.\\n\\n**Lösung**: Pipeline-ID aus Aufrufkontext oder Environment:\\n```python\\nimport os\\n\\ndef get_step_model(step_type: str, pipeline_id: int = None) -> str:\\n if pipeline_id is None:\\n pipeline_id = int(os.environ.get(\\\"PIPELINE_ID\\\", \\\"5\\\"))\\n # ...\\n```\\n\\n### 3.2 ✅ Mehrfachzeilen-Fall absichern\\n\\n**Problem**: Mehrere Zeilen für (pipeline_id, step_type) = Datenfehler.\\n\\n**Lösung A**: UNIQUE Constraint prüfen\/erstellen:\\n```sql\\n-- Prüfen ob existiert:\\nSHOW INDEX FROM pipeline_steps WHERE Column_name = 'step_type';\\n\\n-- Falls nicht: Constraint erstellen\\nALTER TABLE pipeline_steps \\nADD UNIQUE INDEX uq_pipeline_step (pipeline_id, step_type);\\n```\\n\\n**Lösung B**: Im Code absichern:\\n```python\\nrows = cursor.fetchall()\\nif len(rows) > 1:\\n raise PipelineConfigError(\\n f\\\"Datenfehler: Mehrere Einträge für pipeline_id={pipeline_id}, \\\"\\n f\\\"step_type='{step_type}'. UNIQUE Constraint fehlt!\\\"\\n )\\nif len(rows) == 0:\\n raise PipelineConfigError(...)\\n```\\n\\n### 3.3 ✅ Tests erweitern\\n\\n**Zusätzliche Tests (Supervision-Anforderung):**\\n\\n| Test | Zweck |\\n|------|-------|\\n| `test_no_hardcoded_models.py` | Scannt pipeline.py auf Literal-Modelle |\\n| `test_step_config_complete.py` | Prüft ob jeder Step-Call `model` in config hat |\\n| `test_step_type_mapping.py` | Prüft ob step_type Strings korrekt sind (Tippfehler = Ausfall) |\\n\\n### 3.4 ✅ Standalone-Scripts klassifizieren\\n\\n| Script | Nutzung | Entscheidung |\\n|--------|---------|--------------|\\n| `generate_entity_descriptions.py` | Produktiv (Cron) | DB anbinden |\\n| `generate_semantics.py` | Produktiv | DB anbinden |\\n| `backfill_*.py` | Produktiv | ✅ Bereits korrekt |\\n| `quality_test.py` | Dev-only | Explizit ausnehmen |\\n\\n---\\n\\n## 4. KORRIGIERTER LÖSUNGSANSATZ\\n\\n### 4.1 Zentrales Modul: `pipeline_config.py`\\n\\n```python\\n#!\/usr\/bin\/env python3\\n\\\"\\\"\\\"\\nZentrale Pipeline-Konfiguration aus DB.\\n\\nSingle Source of Truth: ki_content.pipeline_steps.config\\n\\\"\\\"\\\"\\n\\nimport os\\nfrom db import db\\n\\n\\nclass PipelineConfigError(Exception):\\n \\\"\\\"\\\"Fehler wenn Config fehlt - KEIN stilles Fallback!\\\"\\\"\\\"\\n pass\\n\\n\\ndef get_step_model(step_type: str, pipeline_id: int = None) -> str:\\n \\\"\\\"\\\"\\n Liest Model aus pipeline_steps.config.\\n \\n Args:\\n step_type: Step-Typ (z.B. 'text_semantic_analyze')\\n pipeline_id: Pipeline-ID (Default aus ENV oder 5)\\n \\n Returns:\\n Model-Name als String\\n \\n Raises:\\n PipelineConfigError: Bei fehlendem\/ungültigem Config\\n \\n WICHTIG:\\n - Wirft Exception wenn nicht konfiguriert!\\n - Kein stilles Fallback auf hardcodierte Werte.\\n - Mehrfachzeilen sind ein Datenfehler.\\n \\\"\\\"\\\"\\n if pipeline_id is None:\\n pipeline_id = int(os.environ.get(\\\"PIPELINE_ID\\\", \\\"5\\\"))\\n \\n cursor = db.execute(\\n \\\"\\\"\\\"SELECT JSON_UNQUOTE(JSON_EXTRACT(config, '$.model')) as model\\n FROM pipeline_steps\\n WHERE pipeline_id = %s AND step_type = %s AND enabled = 1\\\"\\\"\\\",\\n (pipeline_id, step_type)\\n )\\n rows = cursor.fetchall()\\n cursor.close()\\n \\n # Mehrfachzeilen = Datenfehler\\n if len(rows) > 1:\\n raise PipelineConfigError(\\n f\\\"DATENFEHLER: {len(rows)} Einträge für \\\"\\n f\\\"pipeline_id={pipeline_id}, step_type='{step_type}'. \\\"\\n f\\\"Erwarte genau 1. UNIQUE Constraint prüfen!\\\"\\n )\\n \\n # Keine Zeile = Config fehlt\\n if len(rows) == 0:\\n raise PipelineConfigError(\\n f\\\"Kein Model konfiguriert für step_type='{step_type}' \\\"\\n f\\\"in pipeline_id={pipeline_id}. \\\"\\n f\\\"Bitte in pipeline_steps.config setzen!\\\"\\n )\\n \\n model = rows[0].get(\\\"model\\\")\\n \\n # Leerer String = ungültig\\n if not model or model.strip() == \\\"\\\":\\n raise PipelineConfigError(\\n f\\\"Model ist leer für step_type='{step_type}' \\\"\\n f\\\"in pipeline_id={pipeline_id}.\\\"\\n )\\n \\n return model.strip()\\n```\\n\\n### 4.2 Änderungen pro Datei\\n\\n**pipeline.py**\\n```python\\n# Zeile 1: Import hinzufügen\\nfrom pipeline_config import get_step_model\\n\\n# Zeile 242: ALT → NEU\\n# ALT: analyzed_chunks = text_semantic_analyze.execute(chunk_data, {\\\"model\\\": \\\"mistral\\\"})\\nmodel = get_step_model(\\\"text_semantic_analyze\\\")\\nanalyzed_chunks = text_semantic_analyze.execute(chunk_data, {\\\"model\\\": model})\\n\\n# Zeile 286: ALT → NEU\\n# ALT: analyzed_entities = knowledge_semantic_analyze.execute(entity_list, {\\\"model\\\": \\\"mistral\\\"})\\nmodel = get_step_model(\\\"knowledge_semantic_analyze\\\")\\nanalyzed_entities = knowledge_semantic_analyze.execute(entity_list, {\\\"model\\\": model})\\n\\n# Zeile 557: ALT → NEU\\n# ALT: model = \\\"mistral:latest\\\"\\nmodel = get_step_model(\\\"entity_enrich\\\")\\n```\\n\\n**step_semantic_extended.py**\\n```python\\n# Zeile 126 und 286: ALT → NEU\\n# ALT: model = config.get(\\\"model\\\", \\\"mistral\\\")\\nmodel = config.get(\\\"model\\\")\\nif not model:\\n raise ValueError(\\n f\\\"Model muss in config übergeben werden! \\\"\\n f\\\"Step: {self.__class__.__name__}\\\"\\n )\\n```\\n\\n---\\n\\n## 5. DB-VORBEREITUNG\\n\\n### 5.1 UNIQUE Constraint prüfen\\n\\n```sql\\n-- Status prüfen\\nSELECT COUNT(*) as duplicates\\nFROM pipeline_steps\\nGROUP BY pipeline_id, step_type\\nHAVING COUNT(*) > 1;\\n\\n-- Constraint erstellen (falls keine Duplikate)\\nALTER TABLE pipeline_steps \\nADD UNIQUE INDEX uq_pipeline_step (pipeline_id, step_type);\\n```\\n\\n### 5.2 Step-Type für entity_enrich hinzufügen\\n\\n```sql\\n-- Prüfen ob existiert\\nSELECT * FROM pipeline_steps \\nWHERE pipeline_id = 5 AND step_type = 'enrich';\\n\\n-- Falls nicht: Hinzufügen\\nINSERT INTO pipeline_steps (pipeline_id, step_type, config, sort_order, enabled)\\nVALUES (5, 'enrich', '{\\\"model\\\": \\\"mistral\\\"}', 50, 1);\\n```\\n\\n---\\n\\n## 6. ERWEITERTER TEST-PLAN\\n\\n### 6.1 Neue Tests (Supervision-Anforderung)\\n\\n```python\\n# test_no_hardcoded_models.py\\ndef test_pipeline_no_literal_models():\\n \\\"\\\"\\\"Scannt pipeline.py auf hardcodierte Model-Strings.\\\"\\\"\\\"\\n import re\\n with open(\\\"pipeline.py\\\") as f:\\n content = f.read()\\n \\n # Suche nach {\\\"model\\\": \\\"...\\\"} oder model = \\\"...\\\"\\n pattern = r'[\\\"\\\\']model[\\\"\\\\']\\\\s*:\\\\s*[\\\"\\\\'][^\\\"\\\\']+[\\\"\\\\']'\\n matches = re.findall(pattern, content)\\n \\n assert len(matches) == 0, f\\\"Hardcodierte Models gefunden: {matches}\\\"\\n\\n\\n# test_step_type_mapping.py\\ndef test_step_types_exist_in_db():\\n \\\"\\\"\\\"Prüft ob alle verwendeten step_types in DB existieren.\\\"\\\"\\\"\\n from pipeline_config import get_step_model\\n from db import db\\n \\n db.connect()\\n \\n step_types = [\\n \\\"text_semantic_analyze\\\",\\n \\\"knowledge_semantic_analyze\\\",\\n \\\"entity_extract\\\",\\n \\\"enrich\\\",\\n ]\\n \\n for st in step_types:\\n try:\\n model = get_step_model(st)\\n assert model, f\\\"Model leer für {st}\\\"\\n except Exception as e:\\n pytest.fail(f\\\"step_type '{st}' nicht konfiguriert: {e}\\\")\\n \\n db.disconnect()\\n```\\n\\n### 6.2 Integration-Tests\\n\\n- [ ] Pipeline-Run mit DB-Config → verwendet DB-Model\\n- [ ] DB-Update auf anderes Model → nächster Run verwendet neues Model\\n- [ ] Fehlender Config → Exit Code != 0, Fehlermeldung enthält step_type\\n\\n### 6.3 Manuelle Tests\\n\\n- [ ] GUI: Model ändern → Pipeline verwendet neues Model\\n- [ ] CLI: `python pipeline.py process 1` ohne Fehler\\n- [ ] ENV: `PIPELINE_ID=5 python pipeline.py process 1` funktioniert\\n\\n---\\n\\n## 7. RISIKO-BEWERTUNG (aktualisiert)\\n\\n| Risiko | Bewertung | Mitigation |\\n|--------|-----------|------------|\\n| Pipeline-Ausfall bei fehlendem Config | GEWOLLT | Exception macht Konfigfehler sichtbar |\\n| Mehrfachzeilen in DB | MITTEL | UNIQUE Constraint + Code-Prüfung |\\n| Tippfehler in step_type | HOCH | Test `test_step_type_mapping.py` |\\n| Breaking Change CLI-Scripts | MITTEL | Produktive Scripts anbinden, Dev-only ausnehmen |\\... [TRUNCATED-85903e84352851c0]"
}
]
}