diff --git a/generative/db.py b/generative/db.py index 5cd979c..9e791a4 100644 --- a/generative/db.py +++ b/generative/db.py @@ -28,40 +28,48 @@ DB_PATH = Path(os.environ.get("ATOMIC_DB_PATH", _REPO_ROOT / ".cache" / "atomic_analytics.db")) +def _add_column(conn: sqlite3.Connection, table: str, coldef: str) -> None: + """Idempotentes `ALTER TABLE ADD COLUMN `. + + #197 Nachbesserung: „duplicate column name" ist der erwartete No-op (Spalte + existiert bereits). Jeder ANDERE `OperationalError` — insbesondere „database + is locked" bei Parallel-Prozessen — wird NICHT verschluckt, sondern + re-raised. Der frühere pauschale `except OperationalError: pass` deutete + einen Lock-Fehlschlag als „Spalte existiert" fehl: die Spalte fehlte + anschließend, und spätere Inserts crashten. `busy_timeout` (s. init_db) + fängt die Race im Normalfall ab; das re-raise ist die Absicherung. + """ + try: + conn.execute(f"ALTER TABLE {table} ADD COLUMN {coldef}") + except sqlite3.OperationalError as e: + if "duplicate column name" not in str(e).lower(): + raise + + def init_db(path: Path = DB_PATH) -> None: """Erstellt DB + Schema falls nicht vorhanden. Idempotent.""" path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(str(path)) + # #197 Nachbesserung: bei Parallel-Prozessen kann ein ALTER TABLE an einem + # Lock scheitern — busy_timeout lässt SQLite bis 5s auf die Freigabe warten, + # statt sofort mit „database is locked" abzubrechen (kein WAL-Umbau). + conn.execute("PRAGMA busy_timeout=5000") conn.executescript(_SCHEMA) # Migration für bestehende DBs ohne n_dropped - try: - conn.execute("ALTER TABLE pipeline_runs ADD COLUMN n_dropped INT DEFAULT 0") - except sqlite3.OperationalError: - pass - try: - conn.execute("ALTER TABLE pipeline_runs ADD COLUMN n_words INT DEFAULT 0") - except sqlite3.OperationalError: - pass - try: - conn.execute("ALTER TABLE pipeline_runs ADD COLUMN model TEXT DEFAULT ''") - except sqlite3.OperationalError: - pass - try: - conn.execute("ALTER TABLE pipeline_runs ADD COLUMN cost_usd REAL DEFAULT 0.0") - except sqlite3.OperationalError: - pass - try: - conn.execute("ALTER TABLE note_evals ADD COLUMN anchor_rate REAL") - except sqlite3.OperationalError: - pass + _add_column(conn, "pipeline_runs", "n_dropped INT DEFAULT 0") + # #197 Schritt 2: n_extracted = "nach Planner/Extractor generiert" (Funnel-Top). + # Bewusst additiv — n_generated (= geschriebene Notes) bleibt unangetastet, + # damit Alt- und Neu-Zeilen vergleichbar bleiben (keine Migration/Mutation). + _add_column(conn, "pipeline_runs", "n_extracted INT DEFAULT 0") + _add_column(conn, "pipeline_runs", "n_words INT DEFAULT 0") + _add_column(conn, "pipeline_runs", "model TEXT DEFAULT ''") + _add_column(conn, "pipeline_runs", "cost_usd REAL DEFAULT 0.0") + _add_column(conn, "note_evals", "anchor_rate REAL") # Anker-Roh-Counts: für die gepoolte Halluzinationsrate (Σ halluziniert / # Σ gesamt) im Dashboard — die Pipeline berechnet sie ohnehin, persistiert # sie aber bisher nur ins JSONL, nicht in die DB. for _col in ("anchors_total", "anchors_hallucinated"): - try: - conn.execute(f"ALTER TABLE note_evals ADD COLUMN {_col} INT") - except sqlite3.OperationalError: - pass + _add_column(conn, "note_evals", f"{_col} INT") conn.commit() conn.close() @@ -92,21 +100,25 @@ def insert_run(conn: sqlite3.Connection, data: dict) -> None: data-Keys (alle optional ausser run_id): run_id, timestamp, pipeline_version, pdf_source, pdf_key, pdf_label, - n_generated, n_vault, n_inbox, n_merge, n_dropped, n_words, model, - tokens_total, tokens_input, tokens_output, tokens_cache_read, + n_generated, n_extracted, n_vault, n_inbox, n_merge, n_dropped, n_words, + model, tokens_total, tokens_input, tokens_output, tokens_cache_read, duration_s, eval_version + + n_generated = geschriebene Notes (historische Semantik, unangetastet). + n_extracted = "nach Planner/Extractor generiert" (Funnel-Top, #197). Fehlt + der Key (Alt-Aufrufer), wird 0 geschrieben — keine NULL, kein Crash. """ data.setdefault("timestamp", datetime.utcnow().isoformat()) conn.execute( """ INSERT OR REPLACE INTO pipeline_runs (run_id, timestamp, pipeline_version, pdf_source, pdf_key, pdf_label, - n_generated, n_vault, n_inbox, n_merge, n_dropped, n_words, model, + n_generated, n_extracted, n_vault, n_inbox, n_merge, n_dropped, n_words, model, cost_usd, tokens_total, tokens_input, tokens_output, tokens_cache_read, duration_s, eval_version, fully_cached) VALUES (:run_id, :timestamp, :pipeline_version, :pdf_source, :pdf_key, :pdf_label, - :n_generated, :n_vault, :n_inbox, :n_merge, :n_dropped, :n_words, :model, + :n_generated, :n_extracted, :n_vault, :n_inbox, :n_merge, :n_dropped, :n_words, :model, :cost_usd, :tokens_total, :tokens_input, :tokens_output, :tokens_cache_read, :duration_s, :eval_version, :fully_cached) """, @@ -118,6 +130,7 @@ def insert_run(conn: sqlite3.Connection, data: dict) -> None: "pdf_key": data.get("pdf_key"), "pdf_label": data.get("pdf_label"), "n_generated": data.get("n_generated", 0), + "n_extracted": data.get("n_extracted", 0), "n_vault": data.get("n_vault", 0), "n_inbox": data.get("n_inbox", 0), "n_merge": data.get("n_merge", 0), diff --git a/generative/eval_dashboard.py b/generative/eval_dashboard.py index e8374fe..5c6f07d 100644 --- a/generative/eval_dashboard.py +++ b/generative/eval_dashboard.py @@ -447,6 +447,13 @@ def _read_token_runs() -> list[dict]: r = json.loads(line) if r.get("cached"): continue + # #197 Nachbesserung: Bookkeeping-/Event-Records (note_outcome, + # anchor_stats, score_result, stage_outcome …) tragen kein `model` + # (Schema-Invariante, vgl. _is_llm_call_record in eval_dashboard_server). + # Ohne diesen Filter zählen sie als „Calls" → calls wird aufgebläht und + # ein Run mit 0 echten LLM-Calls erschiene als 0-Token-Phantomzeile. + if "model" not in r: + continue tin += r.get("input_tokens", 0) or 0 tout += r.get("output_tokens", 0) or 0 tcr += r.get("cache_read_tokens", 0) or 0 diff --git a/generative/eval_dashboard_server.py b/generative/eval_dashboard_server.py index f87b395..8b205a7 100644 --- a/generative/eval_dashboard_server.py +++ b/generative/eval_dashboard_server.py @@ -414,7 +414,11 @@ def build_data( if jl.exists(): import json as _json2 - for line in jl.read_text(encoding="utf-8", errors="replace").splitlines()[:5]: + # #197 Nachbesserung: stage_outcome-Events stehen am Trace-Anfang + # und tragen kein `model` → ein auf 5 Zeilen gekappter Scan fand das + # model nicht mehr. Fenster auf 50 Zeilen erweitert; Zeilen ohne model + # werden ohnehin übersprungen (break nur bei Treffer). + for line in jl.read_text(encoding="utf-8", errors="replace").splitlines()[:50]: try: m = _json2.loads(line.strip()).get("model", "") if m: diff --git a/generative/orchestrator.py b/generative/orchestrator.py index 66af175..d55ebec 100644 --- a/generative/orchestrator.py +++ b/generative/orchestrator.py @@ -240,6 +240,11 @@ async def _run_with_sem(concept, ctext): ctext = concept_text_window(full_text, search_terms, window_words=400) if not ctext.strip(): print(f" [skip] '{c.title}' nicht im Volltext gefunden (Halluzinations-Schutz)", file=sys.stderr) + # #197 Nachbesserung: bisher stummer Pre-Call-Drop (Konzept nicht im + # Volltext) → Funnel-Event. Konfliktfrei zu #216 (das diesen Block nicht anfasst). + _trace_stage_outcome( + c.title, "extractor", "dropped", drop_reason="empty_extraction", detail="not in fulltext" + ) continue tasks.append(_run_with_sem(c, ctext)) concept_for_idx.append(c.title) @@ -252,9 +257,17 @@ async def _run_with_sem(concept, ctext): for i, r in enumerate(results): if isinstance(r, Exception): print(f" [WARN] Extractor '{concept_for_idx[i]}' fehlgeschlagen: {r}", file=sys.stderr) + # #197 Nachbesserung: harter Call-Ausfall → Funnel-Event. ACHTUNG: + # #216 fügt in genau diesem Zweig `failures.append(...)` hinzu — beim + # Merge beide Zeilen behalten (semantisch unabhängig). + _trace_stage_outcome( + concept_for_idx[i], "extractor", "dropped", drop_reason="call_failed", detail=str(r)[:120] + ) failures.append((concept_for_idx[i], str(r))) elif r is None: - pass # bereits von run_per_concept als [extractor-empty] geloggt + # #197 Nachbesserung: leere Extraktion → Funnel-Event (vorher nur + # von run_per_concept als [extractor-empty] auf stderr geloggt). + _trace_stage_outcome(concept_for_idx[i], "extractor", "dropped", drop_reason="empty_extraction") else: r.refine_key = contexts[i][0].title # plan title als stabiler Fallback-Key (Bug #5) drafts.append(r) @@ -562,6 +575,12 @@ async def _merge_with_sem(members: list[int]) -> AtomicNoteDraft: merged.refine_key = drafts[members[0]].refine_key # plan title für concept_map-Lookup erhalten (Bug #5) cluster_idx_to_merged[members[0]] = merged consumed.update(members[1:]) # nicht-Repräsentanten verwerfen + # #197 Nachbesserung: strukturell derselbe Vorgang wie resolve_sibling_dups + # (ein Draft verschwindet in einen Survivor) — bekam bisher kein Event. + for _k in members[1:]: + _trace_stage_outcome( + drafts[_k].title, "dedup", "dropped", drop_reason="entity_resolution_merge", detail=merged.title + ) print(f" [er-stage4] '{merged.title}' ← {[drafts[k].title for k in members]}", file=sys.stderr) for i, d in enumerate(drafts): @@ -574,6 +593,43 @@ async def _merge_with_sem(members: list[int]) -> AtomicNoteDraft: return result +# --- #197 Schritt 1: Stage-Outcome-Events an den Gate-Punkten --------------- +# EIN konsistentes Trace-Event pro Note, das ihren Weg durch die Gates +# maschinenlesbar macht (Grundlage für den Gate-Funnel, #197 Schritt 3). +# Ergänzt das bestehende Event-Vokabular (note_outcome/score_result/ +# anchor_stats/plan_stats) um genau eine Klasse — kein zweites Tracing-System, +# derselbe trace_event()-Pfad. Verifier und Critic tragen ihr per-Note-Urteil +# bereits via anchor_stats bzw. score_result; dieses Event füllt die bisher +# stummen Gates: Faithfulness (kein Event) und alle Drop-Klassen (Artifact, +# Stage-6-Crash, Exact-/Sibling-Dedup), die vorher nur als aggregierte Zähler +# oder stderr-Prints existierten und aus dem Trace nicht rekonstruierbar waren. +def _trace_stage_outcome( + title: str, + stage: str, + outcome: str, + drop_reason: str | None = None, + detail: str | None = None, +) -> None: + """Schreibt ein `stage_outcome`-Event. + + stage: verifier|critic|faithfulness|dedup|artifact|stage6|… + outcome: passed|downgraded|dropped|skipped + drop_reason: maschinenlesbarer Code (None bei outcome=passed) + detail: optionaler menschenlesbarer Zusatz (z.B. Survivor-Titel, step/phase) + + Lazy Import wie die übrigen orchestrator-Trace-Aufrufe; trace_event greift + zur Laufzeit auf das aktive Backend zu (tests biegen es auf tmp um). + """ + from generative.agents.base import trace_event as _te + + payload: dict = {"title": title, "stage": stage, "outcome": outcome} + if drop_reason is not None: + payload["drop_reason"] = drop_reason + if detail is not None: + payload["detail"] = detail + _te("orchestrator", "stage_outcome", payload) + + # --- Stage-6-Crash-Handling (Issue #17) ------------------------------------ # Eine Note, die in Stage 6 (Verifier/Cross-Reference/Critic) crasht, wird NICHT # als unverifizierter Draft geschrieben, sondern gedroppt + als JSON-Crash-Report @@ -633,9 +689,34 @@ def _collect_stage6_results(results, failed_dir: Path): if isinstance(res, _Stage6Failure): write_crash_report(failed_dir, res.payload) crashes.append(res) + _trace_stage_outcome( + res.payload.get("title", "?"), + "stage6", + "dropped", + drop_reason="stage6_crash", + detail=f"{res.payload.get('step', '?')}/{res.payload.get('phase', '?')}", + ) elif isinstance(res, BaseException): - # Crash außerhalb des guarded Wrappers — defensiv, ohne Payload. - print(f" [WARN] Stage-6 unerwartet fehlgeschlagen (kein Crash-Report): {res}", file=sys.stderr) + # Crash außerhalb des guarded Wrappers — defensiv, ohne per-Note-Payload. + # #197 Nachbesserung: konsistent zum _Stage6Failure-Zweig instrumentieren — + # Crash-Report (statt nur einer stderr-Zeile, damit der Drop diagnostizierbar + # bleibt) + stage_outcome-Event (sonst verschwindet die Note lautlos aus dem + # Funnel). Ohne Payload bleibt der Titel unbekannt ("?"). + print( + f" [WARN] Stage-6 unerwartet fehlgeschlagen (BaseException {type(res).__name__}): {res}", + file=sys.stderr, + ) + write_crash_report( + failed_dir, + { + "title": "?", + "step": "stage6", + "phase": _current_phase(), + "exception": f"{type(res).__name__}: {res}", + "traceback": "".join(traceback.format_exception(type(res), res, res.__traceback__)), + }, + ) + _trace_stage_outcome("?", "stage6", "dropped", drop_reason="stage6_crash", detail=type(res).__name__) else: idx, d = res survived_by_idx[idx] = d @@ -663,6 +744,15 @@ def _apply_faithfulness_gate(draft: AtomicNoteDraft, page_index: dict | None, ci # (realer Hrastinski-E2E 2026-07-05: alle Notes via cross_reference # auf extend gedreht, Gate skippte kommentarlos). print(f" [faithfulness] skipped (action={draft.action})") + _trace_stage_outcome( + draft.title, "faithfulness", "skipped", drop_reason="action_not_create", detail=draft.action + ) + elif ENABLE_FAITHFULNESS_GATE and not page_index and draft.action == "create": + # #197 Nachbesserung: Gate aktiv + create-Note, aber leerer page_index + # (PDF ohne [S. N]-Marker — dokumentierter Fall). Das Gate kann nicht + # greifen; ohne Event wäre dieser Skip im Funnel unsichtbar. + print(" [faithfulness] skipped (kein Page-Index)") + _trace_stage_outcome(draft.title, "faithfulness", "skipped", drop_reason="no_page_index") return from generative.pipeline.faithfulness_gate import run_faithfulness_gate @@ -670,6 +760,11 @@ def _apply_faithfulness_gate(draft: AtomicNoteDraft, page_index: dict | None, ci gate = run_faithfulness_gate(draft.body, page_index, citation) if gate.failed: draft.faithfulness_fail = True + _trace_stage_outcome( + draft.title, "faithfulness", "downgraded", drop_reason="faithfulness_fail", detail=f"{gate.n_failed} failed" + ) + else: + _trace_stage_outcome(draft.title, "faithfulness", "passed", detail=f"{gate.n_supported} supported") for v in gate.verdicts: if v.status.startswith("failed_"): e_txt = f" e={v.entailment:.2f}" if v.entailment is not None else "" @@ -1043,6 +1138,7 @@ def _drop_artifacts(drafts: list[AtomicNoteDraft]) -> list[AtomicNoteDraft]: body_lower = (draft.body or "").lower() if any(phrase in body_lower for phrase in _ABSENCE_PHRASES): dropped.append(draft.title) + _trace_stage_outcome(draft.title, "artifact", "dropped", drop_reason="absence_artifact") else: kept.append(draft) if dropped: @@ -1063,6 +1159,7 @@ def dedup_exact(drafts: list[AtomicNoteDraft], existing_concepts: dict[str, str] for d in drafts: key = _normalize(d.title) if key in seen: + _trace_stage_outcome(d.title, "dedup", "dropped", drop_reason="exact_dup") continue exact_match = existing_concepts.get(d.title.lower().strip()) if exact_match and d.action == "create": @@ -1182,6 +1279,7 @@ def _absorb_alias(name: str) -> None: continue d = drafts[m] drop_idx.add(m) + _trace_stage_outcome(d.title, "dedup", "dropped", drop_reason="sibling_neardup", detail=s.title) for alias in [d.title, *d.aliases]: _absorb_alias(alias) s.source_anchors.extend(d.source_anchors) @@ -2237,6 +2335,12 @@ def main(argv: list[str] | None = None): print(_line, file=sys.stderr) return exit_code + # #197 Schritt 2: Funnel-Top "nach Planner/Extractor generiert" festhalten, + # BEVOR Artifact-/Dedup-/Stage-6-Drops die Liste stutzen. Wird als neues Feld + # n_extracted persistiert; n_generated (= geschriebene Notes) bleibt für + # Alt-Daten-Vergleichbarkeit unverändert. + n_extracted = len(drafts) + # --- Artifact-Detector: Abwesenheits-Noten früh verwerfen (kein LLM-Call) --- drafts = _drop_artifacts(drafts) if not drafts: @@ -2600,6 +2704,7 @@ def main(argv: list[str] | None = None): "pdf_key": source_path.stem.split(" - ")[0].strip().lower(), "pdf_label": source_path.stem.split(" - ")[0].strip(), "n_generated": written, + "n_extracted": n_extracted, "n_vault": vault_count, "n_inbox": inbox_count, "n_merge": sum(1 for d in drafts if getattr(d, "action", "") == "extend"), diff --git a/generative/tests/test_stage_outcome_events.py b/generative/tests/test_stage_outcome_events.py new file mode 100644 index 0000000..b4b236e --- /dev/null +++ b/generative/tests/test_stage_outcome_events.py @@ -0,0 +1,424 @@ +"""Tests für Stage-Outcome-Events an den Gate-Punkten (#197 Schritt 1+2). + +Schritt 1: Ein konsistentes `stage_outcome`-Trace-Event pro Note an jedem +Gate-Punkt, das den Weg durch die Pipeline maschinenlesbar macht (Basis für +den Gate-Funnel, #197 Schritt 3). Deckt die bisher stummen Gates/Drop-Klassen +ab: Artifact-Drop, Stage-6-Crash, Exact-/Sibling-Dedup, Faithfulness. Verifier +und Critic tragen ihr per-Note-Urteil bereits via anchor_stats/score_result. + +Schritt 2: neues DB-Feld `n_extracted` ("nach Planner/Extractor generiert"), +das `n_generated` (= geschriebene Notes, Alt-Daten-Kompatibilität) NICHT +antastet. + +Backend-Muster wie test_per_agent_tracking.py: echtes JsonlBackend auf tmp +umbiegen, JSONL zurücklesen — nie die produktive .cache/runs treffen. +""" + +from __future__ import annotations + +import asyncio +import json +import sqlite3 +from unittest.mock import patch + +import generative.agents.tracing as tracing +from generative.agents.tracing import JsonlBackend +from generative import orchestrator as orch +from generative.pipeline.claims import Claim +from generative.pipeline.faithfulness_gate import ClaimVerdict, GateResult +from generative.schemas.atomic_note import AtomicNoteDraft, ConceptItem, ConceptPlan +from generative.schemas.citation import CitationMeta + + +# --- Helfer ----------------------------------------------------------------- + + +def _draft(**kw) -> AtomicNoteDraft: + base = dict( + title="T", + body="b", + source_anchors=[], + related=[], + tags=[], + synthesis_confidence="low", + action="create", + ) + base.update(kw) + return AtomicNoteDraft(**base) + + +def _capture(monkeypatch, tmp_path): + """Biegt das Trace-Backend auf tmp um; gibt einen Reader für stage_outcome-Events zurück.""" + backend = JsonlBackend(run_dir=tmp_path, run_id="test-run") + monkeypatch.setattr(tracing, "_backend", backend) + + def _read_stage_events() -> list[dict]: + f = tmp_path / "test-run.jsonl" + if not f.exists(): + return [] + events = [json.loads(line) for line in f.read_text(encoding="utf-8").splitlines() if line.strip()] + return [e for e in events if e.get("type") == "stage_outcome"] + + return _read_stage_events + + +def _citation() -> CitationMeta: + return CitationMeta(author="Autor", year="2020", title=None, doi=None, source_file="x.pdf") + + +def _claim(text: str = "Claim-Text (S. 1).") -> Claim: + return Claim(text=text, anchor_page=1, anchor_span=(0, len(text)), risk_types=["number"], is_quote=False) + + +# --- Artifact-Drop ---------------------------------------------------------- + + +def test_drop_artifacts_emits_stage_outcome(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + ghost = _draft(title="Geist-Konzept", body="Dieses Konzept wird im Text nicht behandelt.") + + kept = orch._drop_artifacts([ghost]) + + assert kept == [] + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Geist-Konzept" + assert e["stage"] == "artifact" + assert e["outcome"] == "dropped" + assert e["drop_reason"] == "absence_artifact" + + +def test_drop_artifacts_no_event_for_kept(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + ok = _draft(title="Echtes Konzept", body="Substanzieller Inhalt mit Beleg (S. 3).") + + kept = orch._drop_artifacts([ok]) + + assert [d.title for d in kept] == ["Echtes Konzept"] + assert read() == [] # kein Drop → kein stage_outcome + + +# --- Exact-Dedup ------------------------------------------------------------ + + +def test_dedup_exact_emits_stage_outcome(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + a = _draft(title="Gleicher Titel") + b = _draft(title="Gleicher Titel") # identischer normalisierter Titel → Dup + + kept = orch.dedup_exact([a, b], {}) + + assert len(kept) == 1 + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Gleicher Titel" + assert e["stage"] == "dedup" + assert e["outcome"] == "dropped" + assert e["drop_reason"] == "exact_dup" + + +# --- Sibling-Dedup ---------------------------------------------------------- + + +def test_resolve_sibling_dups_emits_stage_outcome(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + survivor = _draft(title="Alpha", critic_score=4, action="create") + dropped = _draft(title="Beta", critic_score=2, action="extend", extend_path="Alpha") + + kept, n_sib = orch.resolve_sibling_dups([survivor, dropped], {}) + + assert n_sib == 1 + assert [d.title for d in kept] == ["Alpha"] + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Beta" + assert e["stage"] == "dedup" + assert e["outcome"] == "dropped" + assert e["drop_reason"] == "sibling_neardup" + assert e["detail"] == "Alpha" # Survivor-Titel als Detail + + +# --- Stage-6-Crash ---------------------------------------------------------- + + +def test_collect_stage6_results_emits_stage_outcome(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + payload = { + "title": "Crash-Note", + "step": "critic", + "phase": "initial", + "exception": "RuntimeError: boom", + "traceback": "...", + "prompt": "p", + "raw_output": "o", + "draft_body": "b", + "run_meta": {"run_id": "r"}, + } + results = [(0, _draft(title="Good")), orch._Stage6Failure(1, payload)] + + survived, crashes = orch._collect_stage6_results(results, tmp_path / "failed") + + assert [d.title for d in survived] == ["Good"] + assert len(crashes) == 1 + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Crash-Note" + assert e["stage"] == "stage6" + assert e["outcome"] == "dropped" + assert e["drop_reason"] == "stage6_crash" + assert e["detail"] == "critic/initial" # step/phase + + +# --- Faithfulness ----------------------------------------------------------- + + +def test_faithfulness_downgrade_emits_stage_outcome(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + monkeypatch.setattr(orch, "ENABLE_FAITHFULNESS_GATE", True) + verdict = ClaimVerdict(claim=_claim(), status="failed_entailment", evidence="Beleg", entailment=0.1) + fake_gate = GateResult(verdicts=[verdict], failed=True, n_supported=0, n_failed=1, n_abstained=0) + monkeypatch.setattr( + "generative.pipeline.faithfulness_gate.run_faithfulness_gate", + lambda body, page_index, citation, **kw: fake_gate, + ) + + draft = _draft(title="Unbelegt", action="create") + orch._apply_faithfulness_gate(draft, {1: "text"}, _citation()) + + assert draft.faithfulness_fail is True + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Unbelegt" + assert e["stage"] == "faithfulness" + assert e["outcome"] == "downgraded" + assert e["drop_reason"] == "faithfulness_fail" + + +def test_faithfulness_pass_emits_stage_outcome(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + monkeypatch.setattr(orch, "ENABLE_FAITHFULNESS_GATE", True) + fake_gate = GateResult(verdicts=[], failed=False, n_supported=2, n_failed=0, n_abstained=0) + monkeypatch.setattr( + "generative.pipeline.faithfulness_gate.run_faithfulness_gate", + lambda body, page_index, citation, **kw: fake_gate, + ) + + draft = _draft(title="Belegt", action="create") + orch._apply_faithfulness_gate(draft, {1: "text"}, _citation()) + + assert draft.faithfulness_fail is False + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Belegt" + assert e["stage"] == "faithfulness" + assert e["outcome"] == "passed" + assert e.get("drop_reason") is None + + +def test_faithfulness_skip_emits_stage_outcome(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + monkeypatch.setattr(orch, "ENABLE_FAITHFULNESS_GATE", True) + # Gate darf gar nicht laufen (action != create) — trotzdem ein Skip-Event. + monkeypatch.setattr( + "generative.pipeline.faithfulness_gate.run_faithfulness_gate", + lambda *a, **k: (_ for _ in ()).throw(AssertionError("Gate darf nicht laufen")), + ) + + draft = _draft(title="Merge-Stub", action="extend") + orch._apply_faithfulness_gate(draft, {1: "text"}, _citation()) + + events = read() + assert len(events) == 1 + e = events[0] + assert e["stage"] == "faithfulness" + assert e["outcome"] == "skipped" + assert e["drop_reason"] == "action_not_create" + assert e["detail"] == "extend" + + +def test_faithfulness_disabled_emits_no_event(monkeypatch, tmp_path): + read = _capture(monkeypatch, tmp_path) + monkeypatch.setattr(orch, "ENABLE_FAITHFULNESS_GATE", False) + + draft = _draft(title="X", action="create") + orch._apply_faithfulness_gate(draft, {1: "text"}, _citation()) + + assert read() == [] # global deaktiviert → kein Gate, kein Event + + +# --- Nachbesserung #197: bisher stumme Funnel-Lücken ------------------------ + + +def test_entity_resolution_merge_emits_stage_outcome(monkeypatch, tmp_path): + """Cluster-Merge (entity_resolution) verwirft Nicht-Repräsentanten via + `consumed`-Set — strukturell derselbe Vorgang wie sibling-Dedup, bekam aber + kein Event. Der Draft, der aus der finalen Liste verschwindet, muss ein + stage_outcome tragen (Survivor als Detail).""" + read = _capture(monkeypatch, tmp_path) + monkeypatch.setattr(orch, "ENABLE_ENTITY_RESOLUTION", True) + d1 = _draft(title="Red Thread of Information", body="Body Content A") + d2 = _draft(title="Roter Faden der Information", body="Body Content A") + + with patch("generative.orchestrator.embeddings") as mock_emb: + mock_emb.embed_title.side_effect = ["emb1", "emb2"] + mock_emb.cosine.side_effect = [0.95, 0.99] # Title-accept (Stage 1), Body-Cluster (Stage 2) + mock_emb.embed_body.side_effect = ["body_emb1", "body_emb2"] + with patch("generative.orchestrator.canonicalizer.merge_cluster") as mock_merge: + mock_merge.return_value = d1 # Repräsentant überlebt + results = asyncio.run(orch.entity_resolution([d1, d2])) + + assert len(results) == 1 + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Roter Faden der Information" # nicht-Repräsentant, verschwindet + assert e["stage"] == "dedup" + assert e["outcome"] == "dropped" + assert e["drop_reason"] == "entity_resolution_merge" + assert e["detail"] == "Red Thread of Information" # Survivor-Titel + + +def test_collect_stage6_baseexception_emits_event_and_report(monkeypatch, tmp_path): + """Der BaseException-Zweig (z.B. asyncio.CancelledError außerhalb des + guarded Wrappers) droppte die Note bisher lautlos — kein Event, kein + Crash-Report. Der Nachbar-Zweig (_Stage6Failure) hat beides; konsistent + instrumentieren.""" + read = _capture(monkeypatch, tmp_path) + failed_dir = tmp_path / "failed" + results = [(0, _draft(title="Good")), asyncio.CancelledError()] + + survived, crashes = orch._collect_stage6_results(results, failed_dir) + + assert [d.title for d in survived] == ["Good"] + assert crashes == [] # BaseException ist kein _Stage6Failure → nicht in crashes + events = read() + assert len(events) == 1 + e = events[0] + assert e["stage"] == "stage6" + assert e["outcome"] == "dropped" + assert e["drop_reason"] == "stage6_crash" + assert e["detail"] == "CancelledError" + assert list(failed_dir.glob("*.json")) # Crash-Report geschrieben (diagnostizierbar) + + +def _concept(title: str) -> ConceptItem: + return ConceptItem(title=title, priority="high", chapter="Ch", action="create") + + +def test_run_extractors_empty_ctext_emits_stage_outcome(monkeypatch, tmp_path): + """Konzept nicht im Volltext gefunden (`if not ctext.strip(): continue`) — + fiel bisher stumm vor jeder Event-Instrumentierung weg.""" + read = _capture(monkeypatch, tmp_path) + plan = ConceptPlan(source_title="T", source_summary="S", concepts=[_concept("Xyzzy Plughversion Frobnicate")]) + full_text = "Der schnelle braune Fuchs springt über den faulen Hund." # kein Titel-Token enthalten + + asyncio.run(orch.run_extractors_per_concept(full_text, plan, existing_concepts={})) + + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Xyzzy Plughversion Frobnicate" + assert e["stage"] == "extractor" + assert e["outcome"] == "dropped" + assert e["drop_reason"] == "empty_extraction" + + +def test_run_extractors_none_and_exception_emit_stage_outcome(monkeypatch, tmp_path): + """Extractor-Rückgabe None (leer) bzw. Exception (harter Call-Ausfall) + verschwanden bisher nur in stderr + dropped-Zähler.""" + read = _capture(monkeypatch, tmp_path) + + async def fake_run_per_concept(concept, concept_text, existing_concepts, **kw): + if concept.title == "BetaConcept": + raise RuntimeError("boom") + return None # AlphaConcept → leere Extraktion + + monkeypatch.setattr(orch.extractor, "run_per_concept", fake_run_per_concept) + + plan = ConceptPlan( + source_title="T", source_summary="S", concepts=[_concept("AlphaConcept"), _concept("BetaConcept")] + ) + full_text = ( + "Einleitung zum Thema AlphaConcept mit weiterem Kontext und Wörtern. " + "Anschließend Ausführungen zu BetaConcept mit zusätzlichem Fülltext drumherum." + ) + + asyncio.run(orch.run_extractors_per_concept(full_text, plan, existing_concepts={})) + + events = read() + by_reason = {e["drop_reason"]: e for e in events} + assert set(by_reason) == {"empty_extraction", "call_failed"} + assert by_reason["empty_extraction"]["title"] == "AlphaConcept" + assert by_reason["empty_extraction"]["stage"] == "extractor" + assert by_reason["call_failed"]["title"] == "BetaConcept" + assert by_reason["call_failed"]["outcome"] == "dropped" + + +def test_faithfulness_no_page_index_emits_skip(monkeypatch, tmp_path): + """Gate aktiv, action=create, aber leerer page_index (PDF ohne [S. N]-Marker): + griff bisher weder Gate noch Skip-Zweig → gar kein Event.""" + read = _capture(monkeypatch, tmp_path) + monkeypatch.setattr(orch, "ENABLE_FAITHFULNESS_GATE", True) + + draft = _draft(title="Ohne Index", action="create") + orch._apply_faithfulness_gate(draft, {}, _citation()) # page_index leer + + events = read() + assert len(events) == 1 + e = events[0] + assert e["title"] == "Ohne Index" + assert e["stage"] == "faithfulness" + assert e["outcome"] == "skipped" + assert e["drop_reason"] == "no_page_index" + + +# --- Schritt 2: n_extracted-DB-Feld ----------------------------------------- + + +def test_pipeline_runs_has_n_extracted_column(tmp_path): + from generative import db + + path = tmp_path / "test.db" + db.init_db(path) + with db.get_db(path) as conn: + cols = [r[1] for r in conn.execute("PRAGMA table_info(pipeline_runs)").fetchall()] + assert "n_extracted" in cols + + +def test_insert_run_stores_n_extracted_independent_of_n_generated(tmp_path): + from generative import db + + path = tmp_path / "test.db" + db.init_db(path) + with db.get_db(path) as conn: + db.insert_run(conn, {"run_id": "run-1", "n_extracted": 10, "n_generated": 3}) + + conn2 = sqlite3.connect(str(path)) + try: + row = conn2.execute("SELECT n_extracted, n_generated FROM pipeline_runs WHERE run_id='run-1'").fetchone() + finally: + conn2.close() + assert row == (10, 3) # zwei unabhängige Felder, kein Semantik-Overlap + + +def test_insert_run_n_extracted_defaults_zero(tmp_path): + """Alt-Zeilen-Kompatibilität: fehlt n_extracted im Insert, ist es 0 (kein NULL/Crash).""" + from generative import db + + path = tmp_path / "test.db" + db.init_db(path) + with db.get_db(path) as conn: + db.insert_run(conn, {"run_id": "run-legacy", "n_generated": 5}) + + conn2 = sqlite3.connect(str(path)) + try: + row = conn2.execute("SELECT n_extracted FROM pipeline_runs WHERE run_id='run-legacy'").fetchone() + finally: + conn2.close() + assert row == (0,) diff --git a/generative/tests/test_stage_outcome_robustness.py b/generative/tests/test_stage_outcome_robustness.py new file mode 100644 index 0000000..f2f82ee --- /dev/null +++ b/generative/tests/test_stage_outcome_robustness.py @@ -0,0 +1,87 @@ +"""Dashboard-/DB-Robustheit gegenüber den neuen stage_outcome-Events (#197 Nachbesserung). + +stage_outcome-Events landen im selben JSONL-Trace wie echte LLM-Calls, tragen +aber kein `model`-Feld. Ohne Schutz blähen sie den Call-Zähler des Dashboards +auf (`_read_token_runs`) bzw. lassen den model-Backfill des Servers ins Leere +laufen. Zusätzlich: `db._add_column` darf einen Lock-Fehlschlag nicht als +„Spalte existiert" verschlucken. +""" + +from __future__ import annotations + +import json +import sqlite3 + +import pytest + + +def _write_jsonl(runs_dir, name, records): + runs_dir.mkdir(parents=True, exist_ok=True) + (runs_dir / name).write_text("\n".join(json.dumps(r) for r in records), encoding="utf-8") + + +# --- Punkt 5: _read_token_runs zählt nur echte LLM-Calls -------------------- + + +def test_read_token_runs_excludes_event_records(monkeypatch, tmp_path): + from generative import eval_dashboard as ed + + runs = tmp_path / "runs" + _write_jsonl( + runs, + "20260712-101500.jsonl", + [ + {"type": "stage_outcome", "stage": "extractor", "outcome": "dropped"}, # kein model + {"type": "note_outcome", "title": "X"}, # bestehendes Bookkeeping, kein model + {"model": "haiku", "input_tokens": 100, "output_tokens": 50}, + {"model": "opus", "input_tokens": 200, "output_tokens": 80}, + ], + ) + monkeypatch.setattr(ed, "RUNS_DIR", runs) + + result = ed._read_token_runs() + + assert len(result) == 1 + assert result[0]["calls"] == 2 # nur die 2 model-Records, nicht die 2 Events + assert result[0]["tokens_in"] == 300 + assert result[0]["tokens_out"] == 130 + + +def test_read_token_runs_no_phantom_row_for_events_only(monkeypatch, tmp_path): + from generative import eval_dashboard as ed + + runs = tmp_path / "runs" + _write_jsonl( + runs, + "20260712-101500.jsonl", + [{"type": "stage_outcome", "stage": "dedup", "outcome": "dropped"} for _ in range(3)], + ) + monkeypatch.setattr(ed, "RUNS_DIR", runs) + + # Nur Events, 0 echte LLM-Calls → count bleibt 0 → keine 0-Token-Phantomzeile. + assert ed._read_token_runs() == [] + + +# --- Punkt 7: _add_column verschluckt nur duplicate-column ------------------ + + +def test_add_column_swallows_duplicate_but_reraises_other(tmp_path): + from generative import db + + path = tmp_path / "t.db" + db.init_db(path) + conn = sqlite3.connect(str(path)) + try: + # Bestehende Spalte → „duplicate column name" → No-op (kein Raise). + db._add_column(conn, "pipeline_runs", "n_generated INT DEFAULT 0") + # Neue Spalte → wird angelegt; zweiter Aufruf ist ebenfalls No-op. + db._add_column(conn, "pipeline_runs", "brandneu_col INT DEFAULT 0") + db._add_column(conn, "pipeline_runs", "brandneu_col INT DEFAULT 0") + cols = [r[1] for r in conn.execute("PRAGMA table_info(pipeline_runs)").fetchall()] + assert "brandneu_col" in cols + + # Nicht-duplicate OperationalError (kein solcher Tisch) → re-raise, NICHT verschlucken. + with pytest.raises(sqlite3.OperationalError): + db._add_column(conn, "kein_tisch", "x INT") + finally: + conn.close() diff --git a/shared/db_schema.py b/shared/db_schema.py index 779685c..ad1927d 100644 --- a/shared/db_schema.py +++ b/shared/db_schema.py @@ -16,6 +16,7 @@ pdf_key TEXT, pdf_label TEXT, n_generated INT DEFAULT 0, + n_extracted INT DEFAULT 0, n_vault INT DEFAULT 0, n_inbox INT DEFAULT 0, n_merge INT DEFAULT 0,