fix(generative): pipeline_runs-Insert vom Inline-Eval entkoppeln + Test-Traces nach tmp (#198 P1+P2)

generative/conftest.py

@@ -0,0 +1,106 @@
+"""Test-Isolation der Pipeline-Seiteneffekte für ALLE generative-Tests (#198 P2).
+
+Vier produktive Senken wurden bisher aus Tests heraus beschrieben bzw. mutiert:
+
+1. **Trace-Verzeichnis** — jeder LLM-Call (`agents.base.call_claude_full`) und jedes
+   strukturierte Event routet über `agents.tracing._backend` und schreibt eine Zeile
+   nach `generative/.cache/runs/<run-id>.jsonl`. Tests, die `call_claude_full` mit
+   gestubbtem Backend aufrufen (z. B. test_call_record, test_eval_cache_namespace),
+   erzeugten so echte Trace-Dateien mit agent="test"/"verifier"/... im produktiven
+   `runs/`-Verzeichnis.
+
+2. **LLM-Response-Cache** — `call_claude_full(..., use_cache=True)` (Default) legt die
+   Backend-Antwort unter `base._cache_put` in `generative/.cache/llm/<key>.json` ab. Der
+   in-process-main()-Smoke-E2E-Test persistierte damit Fake-Backend-Antworten unter
+   ECHTEN Cache-Keys im produktiven LLM-Cache (#198 Nachbesserung Fix 1).
+
+3. **Cache-Rotation** — `orchestrator.main()` ruft am Lauf-Ende
+   `cache_rotation.rotate_run_caches()` (Default-Verzeichnis `config.CACHE_DIR`, Caps
+   llm=2000 / runs=600, ÄLTESTE Dateien werden GELÖSCHT). Ein in-process-main()-Test auf
+   einer gut gefüllten Maintainer-Maschine löscht so reale Cache-Dateien
+   (#198 Nachbesserung Fix 2).
+
+4. **Analytics-DB** — seit #198 P1 persistiert `orchestrator.main()` den pipeline_run
+   UNBEDINGT (auch ohne Inline-Eval). Damit würde der In-Process-Smoke-E2E-Test
+   (test_ci_smoke_e2e) in die produktive `.cache/atomic_analytics.db` schreiben.
+
+Diese autouse-Fixture biegt alle vier Senken pro Test in ein isoliertes tmp-Verzeichnis
+um — analog zum bestehenden Sicherheitsnetz in `generative/gui/tests/conftest.py`. Bewusst
+über `tmp_path_factory` (eigener Ordner), NICHT über das test-eigene `tmp_path`: viele
+Tests listen ihr `tmp_path` und würden sonst über die eingestreute `analytics.db` /
+`trace-runs/` stolpern. Tests, die selbst einen Trace-Backend oder DB-Pfad setzen,
+überschreiben die Fixture einfach danach (monkeypatch, last-wins) und bleiben unberührt.
+
+Abgedeckt (in-process): tracing._backend, base._RUN_DIR, base._LLM_CACHE_DIR, db.DB_PATH
+(Modul-Attribut) sowie die Lösch-Senke von cache_rotation.rotate_run_caches().
+
+NICHT abgedeckt — bewusst dokumentiert:
+  * **Late-Binding-Falle** — die db-Query-Helfer (`db.query_pipeline_runs`,
+    `query_note_evals`, `query_kpi_trend`, `available_eval_versions`, `init_db`, `get_db`)
+    binden `DB_PATH` als DEFAULT-ARGUMENT zur DEF-Zeit. Ein no-arg-Aufruf (`query_*()`)
+    liest damit die REALE DB und umgeht den `db.DB_PATH`-Monkeypatch dieser Fixture. Nur
+    Aufrufer, die den Pfad explizit durchreichen — wie orchestrator.main() via
+    `get_db(db.DB_PATH)` — sind isoliert. Reine Lese-Query-Aufrufer sind harmlos (kein
+    Write), aber nicht deterministisch gegen die reale DB.
+  * **Subprozess-Tests** — `tests/test_e2e_baseline.py` startet die Pipeline als eigenen
+    Prozess; diese in-process-Monkeypatches greifen dort prinzipiell nicht. Dieser Test
+    isoliert die DB-Senke selbst per `ATOMIC_DB_PATH`-ENV (siehe dortiger Kommentar).
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def isolate_pipeline_side_effects(tmp_path_factory, monkeypatch):
+    """Trace-, LLM-Cache-, Rotations- und DB-Senken landen in tmp, nie im produktiven .cache/.
+
+    Gibt ein SimpleNamespace mit `runs_dir`, `llm_cache_dir`, `cache_root` und `db_path`
+    zurück, damit Tests die Zielpfade abfragen können (per-Namen anfordern, obwohl autouse).
+    """
+    from generative import cache_rotation, db
+    from generative.agents import base, tracing
+
+    base_dir = tmp_path_factory.mktemp("pipeline-side-effects")
+    runs_dir = base_dir / "trace-runs"
+    llm_cache_dir = base_dir / "llm-cache"
+    cache_root = base_dir / "cache-root"
+    db_path = base_dir / "analytics.db"
+
+    # Trace-Backend + Lese-Pfad (base._RUN_DIR, von orchestrator zum Token-Aggregat
+    # genutzt) auf denselben tmp-Ordner zeigen — Schreiben und Lesen konsistent.
+    monkeypatch.setattr(tracing, "_backend", tracing.JsonlBackend(run_dir=runs_dir, run_id=tracing._RUN_ID))
+    monkeypatch.setattr(base, "_RUN_DIR", runs_dir)
+
+    # LLM-Response-Cache (_cache_get/_cache_put) auf tmp umbiegen — Fake-Backend-
+    # Antworten aus Tests dürfen nie unter echten Cache-Keys im produktiven .cache/llm/
+    # landen (Fix 1).
+    monkeypatch.setattr(base, "_LLM_CACHE_DIR", llm_cache_dir)
+
+    # Cache-Rotation (Lösch-Senke): orchestrator.main() ruft rotate_run_caches() ohne
+    # Argument → Default = produktives config.CACHE_DIR, ältestes wird GELÖSCHT. Wrapper
+    # bindet ein tmp-cache_root als Default, damit auf einer vollen Maintainer-Maschine
+    # kein reales .cache/llm bzw. .cache/runs beschnitten wird (Fix 2). Die echte
+    # Rotations-Logik (inkl. Caps) bleibt exerziert — nur das Zielverzeichnis ist tmp.
+    _real_rotate = cache_rotation.rotate_run_caches
+    monkeypatch.setattr(
+        cache_rotation,
+        "rotate_run_caches",
+        lambda cache_dir=cache_root: _real_rotate(cache_dir),
+    )
+
+    # get_db(DB_PATH)/init_db lesen db.DB_PATH zur Laufzeit — Umbiegen genügt.
+    # Schema gleich anlegen: ein Insert (orchestrator #198 P1) findet die Tabelle,
+    # und Queries auf die leere DB liefern [] statt OperationalError.
+    db.init_db(db_path)
+    monkeypatch.setattr(db, "DB_PATH", db_path)
+
+    return SimpleNamespace(
+        runs_dir=runs_dir,
+        llm_cache_dir=llm_cache_dir,
+        cache_root=cache_root,
+        db_path=db_path,
+    )

generative/eval_agent_stats.py

@@ -103,8 +103,16 @@ def run_totals(trace_path: Path) -> dict:
 
     ti = to = tcr = tcc = 0
     cost = 0.0
-    if trace_path.exists():
-        for line in trace_path.read_text(encoding="utf-8").splitlines():
+    # Datei-Lesen defensiv: ein unlesbarer (OSError) oder nicht-UTF-8-Trace
+    # (UnicodeDecodeError) darf die Aggregation nicht werfen — der orchestrator-
+    # Token-Block am Lauf-Ende ruft run_totals ungeschützt auf, ein Crash hier
+    # würde den ganzen Lauf am Ende reißen. Tolerant → Nullen (Contract oben).
+    try:
+        _lines = trace_path.read_text(encoding="utf-8").splitlines() if trace_path.exists() else []
+    except (OSError, UnicodeDecodeError):
+        _lines = []
+    if _lines:
+        for line in _lines:
             if not line.strip():
                 continue
             # Ganze Zeile in try: eine kaputte Zeile (Parse-Fehler, null-Feld,

generative/orchestrator.py

@@ -2485,6 +2485,71 @@ def main(argv: list[str] | None = None):
     except Exception:
         print(f"   -> Zeit:   {_wall_s_early}s  |  Tokens: n/a  |  Quelle: {source_path.name}")
 
+    # --- pipeline_runs-Insert: entkoppelt vom Inline-Eval (#198 P1) ---
+    # Dieser Insert lag früher IN Stage 8. Bei deaktiviertem Inline-Eval (Profil
+    # fast/balanced oder ATOMIC_AGENT_INLINE_EVAL=0) kehrte main() VOR dem Insert
+    # zurück — der Lauf hatte einen vollständigen Trace, aber keine DB-Zeile, war
+    # also keiner Pipeline-Version zuordenbar und fiel aus allen versions-gefilterten
+    # Ansichten. Der Insert hängt an nichts Eval-spezifischem (nur an Trace-Tokens +
+    # Run-Zählern), darum läuft er jetzt unbedingt. Die eval-abhängigen note_evals
+    # werden weiterhin ausschließlich bei aktivem Stage-8-Eval geschrieben.
+    from generative import eval_agent_stats as _eas
+    from generative.agents.base import _RUN_DIR as _run_dir_for_meta
+    from generative.agents.base import _RUN_ID as _run_id_for_meta
+
+    _trace_path = _run_dir_for_meta / f"{_run_id_for_meta}.jsonl"
+    _wall_s = round(_time.time() - _run_start, 1)
+    _pre = _eas.run_totals(_trace_path)  # tolerant bei fehlendem/kaputtem Trace → Nullen
+    _tok_in, _tok_out = _pre["input"], _pre["output"]
+    _tok_cache_r, _tok_cache_c = _pre["cache_read"], _pre["cache_create"]
+    _tok_total = _pre["total"]
+    _cost_usd = _pre["cost_usd"]
+
+    run_meta = {
+        "wall_time_s": _wall_s,
+        "tokens_input": _tok_in,
+        "tokens_output": _tok_out,
+        "tokens_cache_read": _tok_cache_r,
+        "tokens_cache_create": _tok_cache_c,
+        "tokens_total": _tok_total,
+    }
+
+    # DB: pipeline_run persistieren. Eigener try/except — ein DB-Fehler darf den
+    # Lauf nie abbrechen. get_db(DB_PATH) liest den Modul-Pfad zur Laufzeit (wie
+    # calibration.collect), damit Tests ihn auf eine tmp-DB umbiegen statt die
+    # produktive DB zu treffen.
+    try:
+        from generative import config as _db_cfg
+        from generative import db as _db
+        from generative.agents.base import _RUN_ID as _db_run_id
+
+        with _db.get_db(_db.DB_PATH) as _conn:
+            _db.insert_run(
+                _conn,
+                {
+                    "run_id": _db_run_id,
+                    "pipeline_version": AGENT_VERSION,
+                    "pdf_source": source_path.name,
+                    "pdf_key": source_path.stem.split(" - ")[0].strip().lower(),
+                    "pdf_label": source_path.stem.split(" - ")[0].strip(),
+                    "n_generated": written,
+                    "n_vault": vault_count,
+                    "n_inbox": inbox_count,
+                    "n_merge": sum(1 for d in drafts if getattr(d, "action", "") == "extend"),
+                    "n_dropped": dropped_total,
+                    "n_words": word_count,
+                    "model": getattr(_db_cfg, "MODEL_PLANNER", ""),
+                    "cost_usd": _cost_usd,
+                    "tokens_total": _tok_total,
+                    "tokens_input": _tok_in,
+                    "tokens_output": _tok_out,
+                    "tokens_cache_read": _tok_cache_r,
+                    "duration_s": _wall_s,
+                },
+            )
+    except Exception as _db_err:
+        print(f"   [warn] DB-Write fehlgeschlagen: {_db_err}")
+
     # --- Stage 8: Qualitäts-Eval (deterministisch, immer gespeichert) ---
     # Läuft nach jedem Run automatisch — PyMuPDF + Fuzzy + Semantic gegen Quell-PDF.
     # Ergebnisse in .cache/quality_history.jsonl für Longitudinal-Vergleiche.
@@ -2515,62 +2580,10 @@ def main(argv: list[str] | None = None):
                         note_files.append(candidates[0])
                         break
 
-        # Pipeline-Tokens/Kosten (Stages 1–7, vor Eval) für run_meta + DB.
-        # Charakterisiert die Note-Generierung; der Stage-8-Eval-Overhead wird
-        # bewusst NICHT in run_meta/DB attribuiert, nur am Run-Ende geprintet.
-        from generative import eval_agent_stats as _eas
-        from generative.agents.base import _RUN_ID, _RUN_DIR
-
-        _trace_path = _RUN_DIR / f"{_RUN_ID}.jsonl"
-        _wall_s = round(_time.time() - _run_start, 1)
-        _pre = _eas.run_totals(_trace_path)
-        _tok_in, _tok_out = _pre["input"], _pre["output"]
-        _tok_cache_r, _tok_cache_c = _pre["cache_read"], _pre["cache_create"]
-        _tok_total = _pre["total"]
-        _cost_usd = _pre["cost_usd"]
-
-        run_meta = {
-            "wall_time_s": _wall_s,
-            "tokens_input": _tok_in,
-            "tokens_output": _tok_out,
-            "tokens_cache_read": _tok_cache_r,
-            "tokens_cache_create": _tok_cache_c,
-            "tokens_total": _tok_total,
-        }
-
-        # DB: pipeline_run persistieren
-        try:
-            from generative.agents.base import _RUN_ID as _db_run_id
-            from generative import config as _db_cfg
-            from generative import db as _db
-
-            with _db.get_db() as _conn:
-                _db.insert_run(
-                    _conn,
-                    {
-                        "run_id": _db_run_id,
-                        "pipeline_version": AGENT_VERSION,
-                        "pdf_source": source_path.name,
-                        "pdf_key": source_path.stem.split(" - ")[0].strip().lower(),
-                        "pdf_label": source_path.stem.split(" - ")[0].strip(),
-                        "n_generated": written,
-                        "n_vault": vault_count,
-                        "n_inbox": inbox_count,
-                        "n_merge": sum(1 for d in drafts if getattr(d, "action", "") == "extend"),
-                        "n_dropped": dropped_total,
-                        "n_words": word_count,
-                        "model": getattr(_db_cfg, "MODEL_PLANNER", ""),
-                        "cost_usd": _cost_usd,
-                        "tokens_total": _tok_total,
-                        "tokens_input": _tok_in,
-                        "tokens_output": _tok_out,
-                        "tokens_cache_read": _tok_cache_r,
-                        "duration_s": _wall_s,
-                    },
-                )
-        except Exception as _db_err:
-            print(f"   [warn] DB-Write fehlgeschlagen: {_db_err}")
-
+        # run_meta + Pipeline-Tokens/Kosten (_pre, _wall_s, _trace_path) sind oben
+        # bereits berechnet und der pipeline_run bereits persistiert (#198 P1 — der
+        # Insert läuft jetzt unbedingt, auch ohne Inline-Eval). Der Stage-8-Eval-
+        # Overhead bleibt bewusst außerhalb von run_meta/DB, nur am Run-Ende geprintet.
         eval_results, _evaluated_count, _reused_count = run_stage8_eval(
             note_files, source_path, run_meta, fresh_run=bool(getattr(args, "fresh_run", False))
         )

generative/tests/test_cache_isolation.py

@@ -0,0 +1,85 @@
+"""#198 Nachbesserung: LLM-Cache- und Rotations-Senke bleiben tmp-isoliert.
+
+Beweist, dass die autouse-Fixture `isolate_pipeline_side_effects` (generative/conftest.py)
+zwei zuvor ungedeckte produktive Senken abfängt:
+
+  Fix 1 — `base._cache_put` (LLM-Response-Cache): ein `call_claude_full(use_cache=True)`
+          landet in tmp, nie in `<repo>/generative/.cache/llm/`.
+  Fix 2 — `cache_rotation.rotate_run_caches()` (Lösch-Senke): rotiert das tmp-cache_root,
+          nie das produktive `.cache/llm` bzw. `.cache/runs`.
+
+Beide Tests snapshotten das REALE `.cache/`-Verzeichnis rein lesend (glob) und stellen
+sicher, dass die Datei-Menge davor == danach ist — es werden nie echte Cache-Daten
+angefasst. Die "Redirect greift"-Assertion steht bewusst VOR jedem Schreib-/Rotations-
+Schritt: fällt die Isolation aus, bricht der Test ab, bevor irgendetwas passiert.
+"""
+
+from __future__ import annotations
+
+import os
+
+from generative import cache_rotation
+from generative.agents import base
+from generative.agents.base import CallResult
+from generative.config import CACHE_DIR
+
+
+def test_llm_cache_write_goes_to_tmp_not_production(isolate_pipeline_side_effects, monkeypatch):
+    real_llm = CACHE_DIR / "llm"
+    before = set(real_llm.glob("*.json")) if real_llm.exists() else set()
+
+    # Redirect-Nachweis ZUERST — bei ausgefallener Isolation Abbruch vor dem Schreiben.
+    assert base._LLM_CACHE_DIR == isolate_pipeline_side_effects.llm_cache_dir
+    assert base._LLM_CACHE_DIR != real_llm
+
+    # Gestubbtes Backend → deterministischer, netz-freier Call über den echten
+    # _cache_put-Schreibpfad (use_cache=True ist Default).
+    monkeypatch.setattr(
+        base,
+        "_backend_call_full",
+        lambda prompt, **k: CallResult(text="antwort", input_tokens=3, output_tokens=1),
+    )
+    base.call_claude_full("ein prompt", agent="test", use_cache=True)
+
+    # (a) Cache-File liegt in tmp.
+    tmp_files = list(isolate_pipeline_side_effects.llm_cache_dir.glob("*.json"))
+    assert tmp_files, f"kein Cache-File in tmp: {isolate_pipeline_side_effects.llm_cache_dir}"
+
+    # (b) Produktives .cache/llm bekam keine neue Datei.
+    after = set(real_llm.glob("*.json")) if real_llm.exists() else set()
+    assert after == before, f"Produktives .cache/llm verschmutzt: {after - before}"
+
+
+def test_rotation_targets_tmp_not_production(isolate_pipeline_side_effects, monkeypatch):
+    real_llm = CACHE_DIR / "llm"
+    real_runs = CACHE_DIR / "runs"
+    llm_before = set(real_llm.iterdir()) if real_llm.exists() else set()
+    runs_before = set(real_runs.iterdir()) if real_runs.exists() else set()
+
+    # Caps runter, damit Rotation ohne 2001 Dateien greift. rotate_run_caches liest die
+    # Caps als Modul-Globals zur Laufzeit → Monkeypatch wirkt.
+    monkeypatch.setattr(cache_rotation, "CACHE_RUNS_MAX_FILES", 1)
+    monkeypatch.setattr(cache_rotation, "CACHE_LLM_MAX_FILES", 1)
+
+    cache_root = isolate_pipeline_side_effects.cache_root
+    tmp_runs = cache_root / "runs"
+    tmp_runs.mkdir(parents=True, exist_ok=True)
+    old = tmp_runs / "old.jsonl"
+    old.write_text("x", encoding="utf-8")
+    new = tmp_runs / "new.jsonl"
+    new.write_text("y", encoding="utf-8")
+    os.utime(old, (1, 1))  # alt → wird bei keep=1 gelöscht
+
+    # rotate_run_caches() ist von der Fixture auf cache_root umgebogen (kein Argument →
+    # tmp-Default, wie orchestrator.main() es aufruft).
+    _n_llm, n_runs = cache_rotation.rotate_run_caches()
+
+    # (a) Rotation traf das tmp-cache_root: ältestes weg, neuestes bleibt.
+    assert n_runs == 1, f"tmp-runs nicht rotiert (n_runs={n_runs})"
+    assert new.exists() and not old.exists()
+
+    # (b) Produktive .cache/llm und .cache/runs unberührt.
+    llm_after = set(real_llm.iterdir()) if real_llm.exists() else set()
+    runs_after = set(real_runs.iterdir()) if real_runs.exists() else set()
+    assert llm_after == llm_before, f"Produktives .cache/llm verändert: {llm_after ^ llm_before}"
+    assert runs_after == runs_before, f"Produktives .cache/runs verändert: {runs_after ^ runs_before}"

generative/tests/test_e2e_baseline.py

@@ -91,14 +91,24 @@ def parse_output(output: str) -> tuple[int, int]:
 
 @pytest.mark.slow
 @pytest.mark.parametrize("key", ["schlebbe", "kuhlthau", "bates"])
-def test_vault_quote_baseline(key: str):
+def test_vault_quote_baseline(key: str, tmp_path):
     """Vault-Quote darf nicht unter Mindest-Schwelle fallen."""
     pdf = BASELINE_PDFS[key]
     if not pdf.exists():
         pytest.skip(f"Baseline-PDF nicht gefunden: {pdf}")
 
+    # DB-Senke isolieren: dieser Test startet die Pipeline als SUBPROZESS, die
+    # in-process-autouse-Fixture (generative/conftest.py) greift dort NICHT. db.DB_PATH
+    # wird zur db.py-Import-Zeit aus ATOMIC_DB_PATH gelesen — im frischen Prozess also
+    # bindbar. Ohne Override würde jeder Baseline-Lauf eine echte pipeline_runs-Zeile in
+    # die produktive .cache/atomic_analytics.db schreiben (#198 Nachbesserung Fix 3).
+    # Residuum (bewusst NICHT isoliert): Trace (.cache/runs), LLM-Cache (.cache/llm) und
+    # Cache-Rotation kennen keine ENV-Steuerung; sie in tmp umzubiegen bräuchte einen
+    # ATOMIC_CACHE_DIR-Umbau in config.py mit breitem Import-Rippel. Da dies ein ECHTER
+    # Pipeline-Lauf ist (kein Fake-Backend), sind diese Trace-/Cache-Schreibvorgänge
+    # legitime Laufdaten (rotations-gedeckelt), keine schädliche Test-Verschmutzung.
     min_vault, expected_total = THRESHOLDS[key]
-    output = run_pipeline(pdf)
+    output = run_pipeline(pdf, env_overrides={"ATOMIC_DB_PATH": str(tmp_path / "e2e_analytics.db")})
     vault, total = parse_output(output)
 
     assert vault > 0, f"[{key}] Keine Notes extrahiert — Pipeline-Fehler?\n{output[-500:]}"