feat(db): 0-Notes-Läufe in pipeline_runs erfassen (#330)

generative/db.py

@@ -71,6 +71,10 @@ def init_db(path: Path = DB_PATH) -> None:
     # in shared/db_schema.py fuer den Zwei-Phasen-Schreibpfad (insert_run VOR,
     # update_wall_clock_s NACH Stage-8).
     _add_column(conn, "pipeline_runs", "wall_clock_s REAL DEFAULT 0")
+    # #330: 0-Notes-Laeufe (Konzeptmangel/Totalverlust) bekommen jetzt eine
+    # pipeline_runs-Zeile mit Abbruchgrund statt komplett zu fehlen —
+    # additive nullable Spalte, Bestandszeilen bleiben NULL (Erfolgspfad).
+    _add_column(conn, "pipeline_runs", "abort_reason TEXT")
     _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
@@ -109,7 +113,7 @@ def insert_run(conn: sqlite3.Connection, data: dict) -> None:
       run_id, timestamp, pipeline_version, pdf_source, pdf_key, pdf_label,
       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, profile, wall_clock_s
+      duration_s, eval_version, profile, wall_clock_s, abort_reason
 
     n_generated = geschriebene Notes (historische Semantik, unangetastet).
     n_extracted = "nach Planner/Extractor generiert" (Funnel-Top, #197). Fehlt
@@ -119,6 +123,12 @@ def insert_run(conn: sqlite3.Connection, data: dict) -> None:
     der Aufrufer (orchestrator.main()) uebergibt hier bewusst denselben Wert.
     Nach Stage-8 korrigiert update_wall_clock_s() die Zeile auf die echte
     Gesamtzeit inkl. Eval-Phase.
+
+    abort_reason (#330): Grund fuer einen 0-Notes-Lauf (z.B. "no_concepts",
+    "all_secondary_mentions", "all_artifacts", "extraction_total_loss").
+    Fehlt der Key (Erfolgspfad-Aufrufer), bleibt die Spalte NULL — anders als
+    die uebrigen Feldern hier KEIN 0/''-Default, NULL ist die korrekte
+    "kein Abbruch"-Semantik.
     """
     data.setdefault("timestamp", datetime.utcnow().isoformat())
     conn.execute(
@@ -127,12 +137,12 @@ def insert_run(conn: sqlite3.Connection, data: dict) -> None:
           (run_id, timestamp, pipeline_version, pdf_source, pdf_key, pdf_label,
            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, profile, wall_clock_s)
+           duration_s, eval_version, fully_cached, profile, wall_clock_s, abort_reason)
         VALUES
           (:run_id, :timestamp, :pipeline_version, :pdf_source, :pdf_key, :pdf_label,
            :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, :profile, :wall_clock_s)
+           :duration_s, :eval_version, :fully_cached, :profile, :wall_clock_s, :abort_reason)
     """,
         {
             "run_id": data.get("run_id"),
@@ -159,6 +169,7 @@ def insert_run(conn: sqlite3.Connection, data: dict) -> None:
             "fully_cached": 1 if (data.get("tokens_total", 0) == 0 and data.get("duration_s", 0) > 0) else 0,
             "profile": data.get("profile", ""),
             "wall_clock_s": data.get("wall_clock_s", 0.0),
+            "abort_reason": data.get("abort_reason"),
         },
     )
 

generative/orchestrator.py

@@ -422,6 +422,69 @@ def total_loss_warning_line(n_attempted: int) -> str:
     )
 
 
+# --- #330: 0-Notes-Läufe bekommen eine pipeline_runs-Zeile ------------------
+# Läufe, die 0 Notes erzeugen (legitimer Konzeptmangel ODER stiller
+# Totalverlust, #281), kehrten bisher VOR dem Erfolgspfad-Insert (weiter unten
+# in main(), #198 P1) zurück — keine DB-Zeile, also unsichtbar für Run-
+# Zählung, Token-/Kosten-Tracking und die Frage „wie oft produziert die
+# Pipeline nichts?". Diese Funktion schließt die Lücke additiv an beiden
+# betroffenen Rückgabepunkten (s. main()): n_generated/n_vault/n_inbox/
+# n_merge bleiben 0 (keine Note geschrieben), abort_reason macht den Grund
+# sichtbar. Eigenes try/except — ein DB-Fehler darf den Lauf-Exit nie
+# verhindern (gleiches Muster wie der Erfolgspfad-Insert weiter unten).
+def _insert_zero_notes_run(
+    *,
+    run_start: float,
+    source_path: Path,
+    runtime_config,
+    dropped_total: int,
+    word_count: int,
+    n_extracted: int,
+    abort_reason: str,
+) -> None:
+    import time as _time
+
+    from generative import config as _db_cfg
+    from generative import db as _db
+    from generative import eval_agent_stats as _eas
+    from generative.agents.base import _RUN_DIR, _RUN_ID
+
+    try:
+        trace_path = _RUN_DIR / f"{_RUN_ID}.jsonl"
+        wall_s = round(_time.time() - run_start, 1)
+        pre = _eas.run_totals(trace_path)  # tolerant bei fehlendem/kaputtem Trace → Nullen
+        with _db.get_db(_db.DB_PATH) as conn:
+            _db.insert_run(
+                conn,
+                {
+                    "run_id": _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": 0,
+                    "n_extracted": n_extracted,
+                    "n_vault": 0,
+                    "n_inbox": 0,
+                    "n_merge": 0,
+                    "n_dropped": dropped_total,
+                    "n_words": word_count,
+                    "model": getattr(_db_cfg, "MODEL_PLANNER", ""),
+                    "cost_usd": pre["cost_usd"],
+                    "tokens_total": pre["total"],
+                    "tokens_input": pre["input"],
+                    "tokens_output": pre["output"],
+                    "tokens_cache_read": pre["cache_read"],
+                    "duration_s": wall_s,
+                    "profile": runtime_config.profile,
+                    "wall_clock_s": wall_s,
+                    "abort_reason": abort_reason,
+                },
+            )
+    except Exception as _db_err:
+        print(f"   [warn] DB-Write (0-Notes-Lauf) fehlgeschlagen: {_db_err}")
+
+
 def _normalize(title: str) -> str:
     """Normalisiert Titel für Dedup-Vergleich: Kleinbuchstaben, Satzzeichen entfernen."""
     import re
@@ -2625,6 +2688,26 @@ def main(argv: list[str] | None = None):
             print(total_loss_warning_line(n_extract_attempted))
         for _line in format_extractor_failure_report(ctx.extractor_failures, n_extract_attempted):
             print(_line, file=sys.stderr)
+        # #330: 0-Notes-Lauf trotzdem in pipeline_runs erfassen. Drei
+        # unterscheidbare Ursachen: stiller Totalverlust (>=1 Konzept
+        # versucht, 0 überlebt — exit_code bereits _EXIT_TOTAL_LOSS), sonst
+        # legitimer Konzeptmangel — mit Sekundär-Erwähnungen (Planner fand
+        # nur secondary_mention-Konzepte) oder ganz ohne Konzepte/nur Skips.
+        if exit_code == _EXIT_TOTAL_LOSS:
+            _abort_reason = "extraction_total_loss"
+        elif ctx.related_mentions:
+            _abort_reason = "all_secondary_mentions"
+        else:
+            _abort_reason = "no_concepts"
+        _insert_zero_notes_run(
+            run_start=_run_start,
+            source_path=source_path,
+            runtime_config=runtime_config,
+            dropped_total=ctx.dropped_total,
+            word_count=ctx.word_count,
+            n_extracted=0,
+            abort_reason=_abort_reason,
+        )
         return exit_code
 
     # #197 Schritt 2: Funnel-Top "nach Planner/Extractor generiert" festhalten,
@@ -2647,6 +2730,18 @@ def main(argv: list[str] | None = None):
             print(total_loss_warning_line(n_extract_attempted))
         for _line in format_extractor_failure_report(ctx.extractor_failures, n_extract_attempted):
             print(_line, file=sys.stderr)
+        # #330: derselbe 0-Notes-Insert wie oben — hier waren Konzepte bereits
+        # extrahiert (n_extracted >= 1), landeten aber komplett als
+        # Abwesenheits-Artefakte statt echter Notes.
+        _insert_zero_notes_run(
+            run_start=_run_start,
+            source_path=source_path,
+            runtime_config=runtime_config,
+            dropped_total=ctx.dropped_total,
+            word_count=ctx.word_count,
+            n_extracted=n_extracted,
+            abort_reason="all_artifacts",
+        )
         return exit_code
 
     # Qualitäts-Flags aus QualityReport auf alle Notes übertragen

generative/tests/test_db.py

@@ -232,7 +232,12 @@ def test_orchestrator_migrates_existing_db_missing_wall_clock_s_column(tmp_path)
     # Regex statt starrem String-Replace: robust gegen Kommentar-Textaenderungen
     # an der wall_clock_s-Spaltendefinition — matcht Komma + optionale
     # Kommentarzeilen + die Spaltenzeile selbst, unabhaengig vom genauen Wortlaut.
-    old_schema = re.sub(r",\n(\s*--[^\n]*\n)*\s*wall_clock_s\s+REAL DEFAULT 0\n", "\n", SCHEMA_SQL)
+    # #330: wall_clock_s ist seither NICHT mehr die letzte Spalte (abort_reason
+    # folgt) — hat also selbst ein trailing Komma (","?" faengt beide Faelle),
+    # und die Ersetzung muss der VORHERIGEN Spalte ("profile") ihr eigenes
+    # trailing Komma zurueckgeben (",\n" statt nur "\n"), sonst fehlt das
+    # Trennzeichen zu abort_reason und das simulierte Alt-Schema ist ungueltiges SQL.
+    old_schema = re.sub(r",\n(\s*--[^\n]*\n)*\s*wall_clock_s\s+REAL DEFAULT 0,?\n", ",\n", SCHEMA_SQL)
     assert "wall_clock_s" not in old_schema, "Test-Fixture-Bug: Alt-Schema hat wall_clock_s schon"
 
     path = tmp_path / "old-schema.db"
@@ -250,3 +255,96 @@ def test_orchestrator_migrates_existing_db_missing_wall_clock_s_column(tmp_path)
     finally:
         conn2.close()
     assert "wall_clock_s" in cols
+
+
+# --- #330: Abbruchgrund fuer 0-Notes-Laeufe additiv persistieren -----------
+#
+# 0-Notes-Laeufe (Konzeptmangel oder Totalverlust) hinterliessen bisher KEINE
+# pipeline_runs-Zeile. abort_reason ist die additive, nullable Spalte, die den
+# Grund festhaelt, wenn orchestrator.main() so einen Lauf jetzt trotzdem
+# persistiert (n_generated=0). Erfolgspfad-Zeilen bleiben NULL.
+
+
+def test_pipeline_runs_has_abort_reason_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 "abort_reason" in cols
+
+
+def test_insert_run_stores_abort_reason(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": "test-run-zero-notes",
+                "pipeline_version": "v0.0.1",
+                "n_generated": 0,
+                "abort_reason": "no_concepts",
+            },
+        )
+    conn2 = sqlite3.connect(str(path))
+    try:
+        row = conn2.execute("SELECT abort_reason FROM pipeline_runs WHERE run_id='test-run-zero-notes'").fetchone()
+    finally:
+        conn2.close()
+    assert row is not None
+    assert row[0] == "no_concepts"
+
+
+def test_insert_run_abort_reason_defaults_to_null(tmp_path):
+    """Erfolgspfad-Aufrufer uebergeben keinen abort_reason-Key -> muss NULL
+    bleiben, nicht leerer String (unterscheidbar von einem echten Abbruchgrund)."""
+    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": "test-run-success",
+                "pipeline_version": "v0.0.1",
+                "n_generated": 3,
+            },
+        )
+    conn2 = sqlite3.connect(str(path))
+    try:
+        row = conn2.execute("SELECT abort_reason FROM pipeline_runs WHERE run_id='test-run-success'").fetchone()
+    finally:
+        conn2.close()
+    assert row is not None
+    assert row[0] is None
+
+
+def test_orchestrator_migrates_existing_db_missing_abort_reason_column(tmp_path):
+    """Migration wie #235/#239: bestehende DB ohne abort_reason bekommt die
+    Spalte per init_db()-Aufruf nachgezogen (additiv, Bestandszeilen bleiben NULL)."""
+    from generative import db
+    from shared.db_schema import SCHEMA_SQL
+
+    old_schema = re.sub(r",\n(\s*--[^\n]*\n)*\s*abort_reason\s+TEXT\n", "\n", SCHEMA_SQL)
+    assert "abort_reason" not in old_schema, "Test-Fixture-Bug: Alt-Schema hat abort_reason schon"
+
+    path = tmp_path / "old-schema.db"
+    conn = sqlite3.connect(str(path))
+    try:
+        conn.executescript(old_schema)
+        conn.commit()
+    finally:
+        conn.close()
+
+    db.init_db(path)
+    conn2 = sqlite3.connect(str(path))
+    try:
+        cols = [r[1] for r in conn2.execute("PRAGMA table_info(pipeline_runs)").fetchall()]
+    finally:
+        conn2.close()
+    assert "abort_reason" in cols