feat(studio): Phase 2 - Studio-Bausteine, Backend-Kern (Sammeln/Analyse/Faktencheck)

Backend fuer die modularen Studio-Pipeline-Bausteine, alles ueber Claude.

DB (additiv, idempotent):
- Tabellen incident_events + fact_check_runs (+ Indizes)
- Funktion log_incident_event()
- Spalten incidents.summary_at, incidents.executive_summary, articles.geoparsed_at

agents/stage_runners.py (neu): isolierte Bausteine analyze + factcheck auf dem
vorhandenen Bestand, komplett neu, Historie bleibt (Snapshot/fact_check_runs).

orchestrator.py: collect_only in die Lane-Multi-Tenant-Queue eingebaut (enqueue_refresh
4-Tupel, _worker-Entpacken, Multi-Pass-Gate, _run_refresh-Signatur + Analyse/FC-Skip).
Zusaetzlich summary_at beim Schreiben des Lagebilds gestempelt (Studio-Freshness).

incidents.py: Endpunkte POST /{id}/run/{stage} (collect|analyze|factcheck),
GET /{id}/run-status, GET /{id}/freshness, GET /{id}/factcheck-runs[/{run_id}].
Busy-Check auf Online-Orchestrator (_current_tasks), Tenant-Zugriffspruefung.

Kein lokaler LLM-Unterbau, kein lokales Modell. /ask + /events folgen in Phase 3.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/agents/orchestrator.py

@@ -458,9 +458,13 @@ class AgentOrchestrator:
             task.cancel()
         logger.info("Agenten-Orchestrator gestoppt")
 
-    async def enqueue_refresh(self, incident_id: int, trigger_type: str = "manual", user_id: int = None) -> bool:
+    async def enqueue_refresh(self, incident_id: int, trigger_type: str = "manual", user_id: int = None, collect_only: bool = False) -> bool:
         """Refresh-Auftrag in die Lane der Organisation stellen. Gibt False zurueck
-        wenn die Lage dort bereits wartet oder gerade laeuft."""
+        wenn die Lage dort bereits wartet oder gerade laeuft.
+
+        collect_only=True (Studio-Baustein "Sammeln"): sammelt nur Artikel, ueberspringt
+        Analyse und Faktencheck.
+        """
         if incident_id in self._queued_ids or incident_id in self._current_tasks:
             logger.info(f"Refresh fuer Lage {incident_id} uebersprungen: bereits aktiv/in Queue")
             return False
@@ -476,7 +480,7 @@ class AgentOrchestrator:
                 self._lane_workers[lane_key] = asyncio.create_task(self._worker(lane_key, queue))
                 logger.info(f"Neue Worker-Lane fuer Organisation {lane_key} gestartet")
             self._queued_ids.add(incident_id)
-            queue.put_nowait((incident_id, trigger_type, user_id))
+            queue.put_nowait((incident_id, trigger_type, user_id, collect_only))
             queue_size = queue.qsize()
         logger.info(f"Refresh fuer Lage {incident_id} eingereiht (Lane {lane_key}, Queue: {queue_size}, Trigger: {trigger_type})")
 
@@ -582,11 +586,15 @@ class AgentOrchestrator:
                         return
                 continue
 
-            if len(item) == 3:
+            if len(item) == 4:
+                incident_id, trigger_type, user_id, collect_only = item
+            elif len(item) == 3:
                 incident_id, trigger_type, user_id = item
+                collect_only = False
             else:
                 incident_id, trigger_type = item
                 user_id = None
+                collect_only = False
             self._queued_ids.discard(incident_id)
             # Session-Start EINMAL setzen — bleibt ueber Multi-Pass/Retry hinweg stabil
             self._current_tasks[incident_id] = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
@@ -610,14 +618,14 @@ class AgentOrchestrator:
             try:
                 # Research-Lagen: Automatisch 3 Durchläufe nur beim ersten Refresh
                 incident_type, has_summary = await self._get_incident_info(incident_id)
-                use_multi_pass = incident_type == "research" and not has_summary
+                use_multi_pass = incident_type == "research" and not has_summary and not collect_only
 
                 for attempt in range(3):
                     try:
                         if use_multi_pass:
                             await self._run_research_multi_pass(incident_id, trigger_type=trigger_type, user_id=user_id)
                         else:
-                            await self._run_refresh(incident_id, trigger_type=trigger_type, retry_count=attempt, user_id=user_id)
+                            await self._run_refresh(incident_id, trigger_type=trigger_type, retry_count=attempt, user_id=user_id, collect_only=collect_only)
                         last_error = None
                         break  # Erfolg
                     except asyncio.CancelledError:
@@ -783,7 +791,7 @@ class AgentOrchestrator:
             await db.close()
         return visibility, created_by, tenant_id
 
-    async def _run_refresh(self, incident_id: int, trigger_type: str = "manual", retry_count: int = 0, user_id: int = None, _suppress_complete: bool = False, _pass_info: dict = None):
+    async def _run_refresh(self, incident_id: int, trigger_type: str = "manual", retry_count: int = 0, user_id: int = None, _suppress_complete: bool = False, _pass_info: dict = None, collect_only: bool = False):
         """Führt einen kompletten Refresh-Zyklus durch."""
         import aiosqlite
         from database import get_db
@@ -1433,7 +1441,8 @@ class AgentOrchestrator:
                     logger.warning(f"Quellen-Statistiken konnten nicht aktualisiert werden: {e}")
 
             # Schritt 3+4: Analyse und Faktencheck PARALLEL
-            if new_count > 0 or not previous_summary:
+            # collect_only (Studio-Baustein "Sammeln"): ueberspringt Analyse/Faktencheck.
+            if (new_count > 0 or not previous_summary) and not collect_only:
                 is_first_summary = not previous_summary
 
                 # Snapshot des alten Lagebilds sichern BEVOR parallele Verarbeitung startet
@@ -1775,9 +1784,14 @@ class AgentOrchestrator:
                                     pass
                     sources_json = json.dumps(sources, ensure_ascii=False) if sources else previous_sources_json
 
+                    # summary_at haelt fest, WANN das Lagebild entstand (Studio-Freshness).
+                    # JETZT stempeln, nicht 'now' vom Lauf-Beginn: sonst saehen die in diesem
+                    # Lauf gesammelten Artikel neuer aus als der Bericht, der sie schon enthaelt.
+                    summary_now = datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S')
                     await db.execute(
-                        "UPDATE incidents SET summary = ?, sources_json = ?, executive_summary = NULL, updated_at = ? WHERE id = ?",
-                        (new_summary, sources_json, now, incident_id),
+                        "UPDATE incidents SET summary = ?, sources_json = ?, executive_summary = NULL, "
+                        "updated_at = ?, summary_at = ? WHERE id = ?",
+                        (new_summary, sources_json, summary_now, summary_now, incident_id),
                     )
 
                     # Beim ersten Refresh: Snapshot des neuen Lagebilds erstellen

src/agents/stage_runners.py

@@ -0,0 +1,257 @@
+"""
+Modulare Pipeline-Bausteine fuers Studio.
+
+Fuehrt einzelne Pipeline-Stufen ISOLIERT auf dem vorhandenen DB-Bestand aus,
+statt des durchlaufenden orchestrator._run_refresh. Das grosse _run_refresh
+bleibt unangetastet (dient weiter als "kompletter Lauf").
+
+Nutzt die bereits reinen Agenten (AnalyzerAgent/FactCheckerAgent) und bildet nur
+die Lade-/Persistenz-Logik aus _run_refresh nach.
+
+Verhalten: KOMPLETT NEU (ersetzt das aktive Ergebnis), Historie bleibt einsehbar:
+  - Analyse     -> altes Lagebild wird als incident_snapshots archiviert, dann neu gesetzt
+  - Faktencheck -> alter Faktenstand wird als fact_check_runs archiviert, dann ersetzt
+
+Phase 1: analyze, factcheck. (Spaeter: collect, geoparse, network.)
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from datetime import datetime
+
+from config import TIMEZONE
+from database import get_db
+
+logger = logging.getLogger("osint.stages")
+
+# Im Studio einzeln startbare Bausteine (Phase 1)
+STAGES = {"analyze", "factcheck"}
+
+# Status-Registry je Incident (Single-Flight) + Task-Referenzen (GC-Schutz)
+_STATE: dict[int, dict] = {}
+_TASKS: set = set()
+
+
+def _now() -> str:
+    return datetime.now(TIMEZONE).strftime("%Y-%m-%d %H:%M:%S")
+
+
+def get_state(incident_id: int) -> dict | None:
+    return _STATE.get(incident_id)
+
+
+def is_running(incident_id: int) -> bool:
+    st = _STATE.get(incident_id)
+    return bool(st and st.get("status") == "running")
+
+
+async def _load_incident(db, incident_id: int) -> dict | None:
+    cur = await db.execute("SELECT * FROM incidents WHERE id = ?", (incident_id,))
+    row = await cur.fetchone()
+    return dict(row) if row else None
+
+
+async def _output_language(db, tenant_id) -> str:
+    from services.org_settings import get_org_language, language_display
+    iso = await get_org_language(db, tenant_id) if tenant_id else "de"
+    return language_display(iso)
+
+
+async def _count(db, sql: str, params) -> int:
+    row = await (await db.execute(sql, params)).fetchone()
+    return (row[0] if row else 0) or 0
+
+
+# ---------------------------------------------------------------------------
+# Baustein: Analyse (Lagebild / Recherchebericht) — komplett neu
+# ---------------------------------------------------------------------------
+async def _run_analysis(db, inc: dict, user_id) -> dict:
+    from agents.analyzer import AnalyzerAgent, build_fact_context_block
+    from services.license_service import charge_usage_to_tenant
+
+    incident_id = inc["id"]
+    tenant_id = inc.get("tenant_id")
+    incident_type = inc.get("type") or "adhoc"
+    title = inc.get("title") or ""
+    description = inc.get("description") or ""
+    prev_summary = inc.get("summary") or ""
+    prev_sources = inc.get("sources_json")
+    now = _now()
+    output_language = await _output_language(db, tenant_id)
+
+    # 1) Altes Lagebild als Snapshot archivieren (Historie), bevor es ersetzt wird
+    if prev_summary:
+        acnt = await _count(db, "SELECT COUNT(*) FROM articles WHERE incident_id = ?", (incident_id,))
+        fcnt = await _count(db, "SELECT COUNT(*) FROM fact_checks WHERE incident_id = ?", (incident_id,))
+        await db.execute(
+            """INSERT INTO incident_snapshots
+               (incident_id, summary, sources_json, article_count, fact_check_count,
+                refresh_log_id, created_at, tenant_id)
+               VALUES (?,?,?,?,?,?,?,?)""",
+            (incident_id, prev_summary, prev_sources, acnt, fcnt, None, now, tenant_id),
+        )
+        await db.commit()
+
+    # 2) Alle Artikel + bestehende Fakten laden (Frische-Bias bei adhoc)
+    order = "published_at IS NULL, published_at DESC" if incident_type == "adhoc" else "collected_at DESC"
+    arts = [dict(r) for r in await (await db.execute(
+        f"SELECT * FROM articles WHERE incident_id = ? ORDER BY {order}", (incident_id,)
+    )).fetchall()]
+    if not arts:
+        raise ValueError("Keine Artikel vorhanden — bitte zuerst sammeln.")
+
+    facts = [dict(r) for r in await (await db.execute(
+        "SELECT id, claim, status, sources_count, evidence FROM fact_checks WHERE incident_id = ?",
+        (incident_id,),
+    )).fetchall()]
+    fact_ctx = ""
+    try:
+        fact_ctx = build_fact_context_block(facts, [], incident_type)
+    except Exception as e:
+        logger.warning(f"Faktenkontext fuer Analyse fehlgeschlagen: {e}")
+
+    # 3) Analyse komplett neu ueber ALLE Artikel
+    analyzer = AnalyzerAgent()
+    analysis, usage = await analyzer.analyze(
+        title, description, arts, incident_type,
+        fact_context_block=fact_ctx, output_language=output_language,
+    )
+    if usage:
+        try:
+            await charge_usage_to_tenant(db, tenant_id, usage, source="analysis")
+        except Exception:
+            pass
+    if not analysis or not (analysis.get("summary") or "").strip():
+        raise ValueError("Analyse lieferte kein Lagebild.")
+
+    summary = analysis.get("summary") or ""
+    sources = analysis.get("sources") or []
+    for s in sources:
+        if isinstance(s.get("nr"), str):
+            try:
+                s["nr"] = int(s["nr"])
+            except ValueError:
+                pass
+    sources_json = json.dumps(sources, ensure_ascii=False) if sources else prev_sources
+
+    # summary_at = Entstehungszeit des Lagebilds. JETZT stempeln, nicht 'now' vom Beginn
+    # des Bausteins: die Analyse laeuft Minuten, und der Stempel muss den Stand abdecken,
+    # der tatsaechlich verarbeitet wurde. (updated_at wird auch beim Sammeln gesetzt und
+    # taugt als Bericht-Zeitpunkt ohnehin nicht.)
+    summary_now = _now()
+    await db.execute(
+        "UPDATE incidents SET summary = ?, sources_json = ?, executive_summary = NULL, "
+        "updated_at = ?, summary_at = ? WHERE id = ?",
+        (summary, sources_json, summary_now, summary_now, incident_id),
+    )
+    await db.commit()
+    return {"articles": len(arts), "summary_len": len(summary), "sources": len(sources)}
+
+
+# ---------------------------------------------------------------------------
+# Baustein: Faktencheck — komplett neu (alter Stand als Lauf archiviert)
+# ---------------------------------------------------------------------------
+async def _run_factcheck(db, inc: dict, user_id) -> dict:
+    from agents.factchecker import FactCheckerAgent, deduplicate_new_facts
+    from services.license_service import charge_usage_to_tenant
+
+    incident_id = inc["id"]
+    tenant_id = inc.get("tenant_id")
+    incident_type = inc.get("type") or "adhoc"
+    title = inc.get("title") or ""
+    now = _now()
+    output_language = await _output_language(db, tenant_id)
+
+    # 1) Aktuellen Faktenstand als Lauf archivieren (Historie)
+    cur_facts = [dict(r) for r in await (await db.execute(
+        "SELECT claim, status, sources_count, evidence, is_notification, checked_at, status_history "
+        "FROM fact_checks WHERE incident_id = ? ORDER BY id", (incident_id,)
+    )).fetchall()]
+    if cur_facts:
+        await db.execute(
+            "INSERT INTO fact_check_runs (incident_id, tenant_id, created_at, facts_json, fact_count) VALUES (?,?,?,?,?)",
+            (incident_id, tenant_id, now, json.dumps(cur_facts, ensure_ascii=False), len(cur_facts)),
+        )
+        await db.commit()
+
+    # 2) Alle Artikel laden
+    arts = [dict(r) for r in await (await db.execute(
+        "SELECT * FROM articles WHERE incident_id = ? ORDER BY collected_at DESC", (incident_id,)
+    )).fetchall()]
+    if not arts:
+        raise ValueError("Keine Artikel vorhanden — bitte zuerst sammeln.")
+
+    # 3) Faktencheck komplett neu
+    fc = FactCheckerAgent()
+    facts, usage = await fc.check(title, arts, incident_type, output_language=output_language)
+    if usage:
+        try:
+            await charge_usage_to_tenant(db, tenant_id, usage, source="factcheck")
+        except Exception:
+            pass
+    facts = deduplicate_new_facts(facts or [])
+
+    # 4) Bestehende Fakten ersetzen
+    await db.execute("DELETE FROM fact_checks WHERE incident_id = ?", (incident_id,))
+    for f in facts:
+        init_hist = json.dumps([{"status": f.get("status", "developing"), "at": now}])
+        await db.execute(
+            """INSERT INTO fact_checks
+               (incident_id, claim, status, sources_count, evidence, is_notification, tenant_id, status_history, checked_at)
+               VALUES (?,?,?,?,?,?,?,?,?)""",
+            (incident_id, f.get("claim", ""), f.get("status", "developing"),
+             f.get("sources_count", 0), f.get("evidence"), f.get("is_notification", 0),
+             tenant_id, init_hist, now),
+        )
+    await db.commit()
+    return {"facts": len(facts), "archived": len(cur_facts), "articles": len(arts)}
+
+
+# ---------------------------------------------------------------------------
+# Orchestrierung: Start (Single-Flight) + Status
+# ---------------------------------------------------------------------------
+_RUNNERS = {"analyze": _run_analysis, "factcheck": _run_factcheck}
+
+_LABELS = {"analyze": "Analyse", "factcheck": "Faktencheck"}
+
+
+def start_stage(incident_id: int, stage: str, user_id) -> bool:
+    """Startet einen Baustein im Hintergrund. False, wenn schon einer laeuft."""
+    if stage not in STAGES:
+        raise ValueError(f"Unbekannter Baustein: {stage}")
+    if is_running(incident_id):
+        return False
+    _STATE[incident_id] = {
+        "stage": stage, "label": _LABELS.get(stage, stage),
+        "status": "running", "started_at": _now(),
+        "finished_at": None, "error": None, "result": None,
+    }
+    t = asyncio.create_task(_execute(incident_id, stage, user_id))
+    _TASKS.add(t)
+    t.add_done_callback(_TASKS.discard)
+    return True
+
+
+async def _execute(incident_id: int, stage: str, user_id):
+    db = await get_db()
+    started_at = _STATE.get(incident_id, {}).get("started_at")
+    try:
+        inc = await _load_incident(db, incident_id)
+        if not inc:
+            raise ValueError("Lage nicht gefunden")
+        res = await _RUNNERS[stage](db, inc, user_id)
+        _STATE[incident_id] = {
+            "stage": stage, "label": _LABELS.get(stage, stage), "status": "done",
+            "started_at": started_at, "finished_at": _now(), "error": None, "result": res,
+        }
+        logger.info(f"Baustein {stage} Lage {incident_id} fertig: {res}")
+    except Exception as e:
+        logger.warning(f"Baustein {stage} Lage {incident_id} Fehler: {e}", exc_info=True)
+        _STATE[incident_id] = {
+            "stage": stage, "label": _LABELS.get(stage, stage), "status": "error",
+            "started_at": started_at, "finished_at": _now(), "error": str(e)[:400], "result": None,
+        }
+    finally:
+        await db.close()

src/database.py

@@ -1,5 +1,6 @@
 """SQLite Datenbank-Setup und Zugriff."""
 import aiosqlite
+import json
 import logging
 import os
 from config import DB_PATH, DATA_DIR
@@ -133,6 +134,32 @@ CREATE TABLE IF NOT EXISTS refresh_pipeline_steps (
 CREATE INDEX IF NOT EXISTS idx_pipeline_steps_incident ON refresh_pipeline_steps(incident_id, started_at DESC);
 CREATE INDEX IF NOT EXISTS idx_pipeline_steps_log ON refresh_pipeline_steps(refresh_log_id);
 
+-- Aktivitaets-/Ereignisprotokoll einer Lage (Studio-Ereignis-Timeline).
+-- Erfasst nur Ereignisse, die sonst nirgends stehen: Chat-Q&A + Quellen-Aenderungen.
+CREATE TABLE IF NOT EXISTS incident_events (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    incident_id INTEGER REFERENCES incidents(id) ON DELETE CASCADE,
+    event_type TEXT NOT NULL,          -- 'chat_qa' | 'source_change'
+    title TEXT,
+    detail TEXT,
+    meta TEXT,                         -- optionales JSON
+    user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    tenant_id INTEGER REFERENCES organizations(id)
+);
+CREATE INDEX IF NOT EXISTS idx_incident_events ON incident_events(incident_id, created_at DESC);
+
+-- Archivierte Faktencheck-Laeufe (Studio-Faktencheck-Verlauf).
+CREATE TABLE IF NOT EXISTS fact_check_runs (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    incident_id INTEGER REFERENCES incidents(id) ON DELETE CASCADE,
+    tenant_id INTEGER REFERENCES organizations(id),
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    facts_json TEXT,
+    fact_count INTEGER DEFAULT 0
+);
+CREATE INDEX IF NOT EXISTS idx_fact_check_runs ON fact_check_runs(incident_id, created_at DESC);
+
 CREATE TABLE IF NOT EXISTS incident_snapshots (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     incident_id INTEGER REFERENCES incidents(id) ON DELETE CASCADE,
@@ -369,6 +396,28 @@ async def get_db() -> aiosqlite.Connection:
     return db
 
 
+async def log_incident_event(db, incident_id, event_type, title, detail=None,
+                             meta=None, user_id=None, tenant_id=None, commit=True):
+    """Schreibt ein Ereignis ins Aktivitaetsprotokoll einer Lage (Studio-Timeline).
+
+    Bewusst tolerant: Fehler beim Protokollieren duerfen die eigentliche Aktion
+    (Chat-Antwort, Quelle anlegen) niemals scheitern lassen.
+    """
+    try:
+        await db.execute(
+            """INSERT INTO incident_events
+               (incident_id, event_type, title, detail, meta, user_id, tenant_id)
+               VALUES (?, ?, ?, ?, ?, ?, ?)""",
+            (incident_id, event_type, title, detail,
+             json.dumps(meta, ensure_ascii=False) if meta is not None else None,
+             user_id, tenant_id),
+        )
+        if commit:
+            await db.commit()
+    except Exception as e:  # pragma: no cover - Protokoll ist nie kritisch
+        logger.warning(f"incident_event nicht protokolliert (incident={incident_id}, typ={event_type}): {e}")
+
+
 async def init_db():
     """Initialisiert die Datenbank mit dem Schema."""
     db = await get_db()
@@ -444,6 +493,24 @@ async def init_db():
             await db.commit()
             logger.info("Migration: public_mood_updated_at zu incidents hinzugefuegt")
 
+        # Migration (Studio): summary_at = Entstehungszeit des Lagebilds. Grundlage der
+        # Studio-Anzeige "N neue Artikel seit dem letzten Bericht". updated_at taugt dafuer
+        # nicht (wird auch beim reinen Sammeln gesetzt). Backfill mit updated_at genuegt.
+        if "summary_at" not in columns:
+            await db.execute("ALTER TABLE incidents ADD COLUMN summary_at TEXT")
+            await db.execute(
+                "UPDATE incidents SET summary_at = updated_at "
+                "WHERE summary IS NOT NULL AND TRIM(summary) <> ''"
+            )
+            await db.commit()
+            logger.info("Migration: summary_at zu incidents hinzugefuegt (Studio)")
+
+        # Migration (Studio): executive_summary (stage_runners setzt es beim Analyse-Baustein)
+        if "executive_summary" not in columns:
+            await db.execute("ALTER TABLE incidents ADD COLUMN executive_summary TEXT")
+            await db.commit()
+            logger.info("Migration: executive_summary zu incidents hinzugefuegt (Studio)")
+
         # Migration: Tabelle podcast_transcripts (URL-Cache fuer Transkripte)
         cursor = await db.execute(
             "SELECT name FROM sqlite_master WHERE type='table' AND name='podcast_transcripts'"
@@ -606,6 +673,17 @@ async def init_db():
             await db.execute("ALTER TABLE articles ADD COLUMN tenant_id INTEGER REFERENCES organizations(id)")
             await db.commit()
 
+        # Migration (Studio): geoparsed_at fuer articles (Merker "schon verortet",
+        # Grundlage der Studio-Freshness fuer Geoparsing). Backfill aus article_locations.
+        if "geoparsed_at" not in art_columns:
+            await db.execute("ALTER TABLE articles ADD COLUMN geoparsed_at TEXT")
+            await db.execute(
+                """UPDATE articles SET geoparsed_at = COALESCE(collected_at, CURRENT_TIMESTAMP)
+                   WHERE id IN (SELECT DISTINCT article_id FROM article_locations)"""
+            )
+            await db.commit()
+            logger.info("Migration: geoparsed_at zu articles hinzugefuegt (Studio)")
+
         # Migration: tenant_id fuer fact_checks
         cursor = await db.execute("PRAGMA table_info(fact_checks)")
         fc_columns = [row[1] for row in await cursor.fetchall()]

src/routers/incidents.py

@@ -5,7 +5,7 @@ from models import IncidentCreate, IncidentUpdate, IncidentResponse, IncidentLis
 from auth import get_current_user
 from middleware.license_check import require_writable_license
 from database import db_dependency, get_db
-from datetime import datetime
+from datetime import datetime, timezone
 from config import TIMEZONE
 import asyncio
 import aiosqlite
@@ -1128,6 +1128,184 @@ async def cancel_refresh(
     return {"status": "cancelling" if cancelled else "not_running"}
 
 
+# --- Modulare Pipeline-Bausteine (nur Studio): einzelne Stufen isoliert ------
+@router.post("/{incident_id}/run/{stage}")
+async def run_stage(
+    incident_id: int,
+    stage: str,
+    current_user: dict = Depends(require_writable_license),
+    db: aiosqlite.Connection = Depends(db_dependency),
+):
+    """Einzelnen Pipeline-Baustein auf dem vorhandenen Datenbestand starten
+    (collect | analyze | factcheck). Komplett neu; Historie bleibt erhalten."""
+    from agents import stage_runners
+    from agents.orchestrator import orchestrator
+    user_id = current_user["id"]
+    tenant_id = current_user.get("tenant_id")
+    await _check_incident_access(db, incident_id, user_id, tenant_id)
+
+    orch_busy = (incident_id in getattr(orchestrator, "_current_tasks", {})) or \
+                (incident_id in getattr(orchestrator, "_queued_ids", set()))
+    busy_msg = "Es laeuft bereits eine Aktualisierung/ein Baustein fuer diese Lage."
+
+    # "Sammeln" nutzt die bewaehrte Sammel-Pipeline (Orchestrator, collect_only)
+    if stage == "collect":
+        if stage_runners.is_running(incident_id):
+            raise HTTPException(status_code=409, detail=busy_msg)
+        ok = await orchestrator.enqueue_refresh(
+            incident_id, trigger_type="collect", user_id=user_id, collect_only=True)
+        if not ok:
+            raise HTTPException(status_code=409, detail=busy_msg)
+        return {"started": True, "stage": "collect", "via": "refresh"}
+
+    # Analyse/Faktencheck: isolierte Bausteine (stage_runners)
+    if stage not in stage_runners.STAGES:
+        raise HTTPException(status_code=400, detail=f"Unbekannter Baustein: {stage}")
+    if orch_busy:
+        raise HTTPException(status_code=409, detail=busy_msg)
+    started = stage_runners.start_stage(incident_id, stage, user_id)
+    if not started:
+        raise HTTPException(status_code=409, detail=busy_msg)
+    return {"started": True, "stage": stage}
+
+
+def _app_ts_to_utc(ts) -> str | None:
+    """App-Zeitstempel (lokale Zeitzone) -> UTC-String, fuer den Vergleich mit
+    articles.collected_at (das SQLite in UTC setzt). Ohne Umrechnung waere der
+    Vergleich im Sommer zwei Stunden falsch."""
+    if not ts:
+        return None
+    s = str(ts).strip().replace("T", " ")[:19]
+    try:
+        dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
+    except ValueError:
+        return None
+    return dt.replace(tzinfo=TIMEZONE).astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
+
+@router.get("/{incident_id}/freshness")
+async def get_freshness(
+    incident_id: int,
+    current_user: dict = Depends(get_current_user),
+    db: aiosqlite.Connection = Depends(db_dependency),
+):
+    """Datenstand je Artefakt: erzeugt? wie alt? wie viele Artikel kamen seither dazu?
+
+    Grundlage fuer die Veraltet-Anzeige der Studio-Karten ("12 neue Artikel seit dem
+    letzten Lagebild").
+    """
+    await _check_incident_access(db, incident_id, current_user["id"], current_user.get("tenant_id"))
+
+    async def one(sql: str, params=()):
+        row = await (await db.execute(sql, params)).fetchone()
+        return row[0] if row else None
+
+    async def newer_than(app_ts) -> int:
+        utc = _app_ts_to_utc(app_ts)
+        if not utc:
+            return 0
+        return await one(
+            "SELECT COUNT(*) FROM articles WHERE incident_id = ? AND collected_at > ?",
+            (incident_id, utc),
+        ) or 0
+
+    articles = await one("SELECT COUNT(*) FROM articles WHERE incident_id = ?", (incident_id,)) or 0
+
+    inc = await (await db.execute(
+        "SELECT summary, summary_at, updated_at FROM incidents WHERE id = ?", (incident_id,)
+    )).fetchone()
+    has_summary = bool(inc and (inc["summary"] or "").strip())
+    # summary_at ist die Entstehungszeit des Lagebilds; updated_at nur der Notnagel.
+    summary_at = (inc["summary_at"] or inc["updated_at"]) if inc else None
+
+    facts = await one("SELECT COUNT(*) FROM fact_checks WHERE incident_id = ?", (incident_id,)) or 0
+    fc_at = await one("SELECT MAX(checked_at) FROM fact_checks WHERE incident_id = ?", (incident_id,))
+
+    geo_pending = await one(
+        "SELECT COUNT(*) FROM articles WHERE incident_id = ? AND geoparsed_at IS NULL",
+        (incident_id,),
+    ) or 0
+    geo_at = await one(
+        "SELECT MAX(geoparsed_at) FROM articles WHERE incident_id = ?", (incident_id,)
+    )
+
+    return {
+        "articles": articles,
+        "summary": {
+            "exists": has_summary,
+            "last": summary_at if has_summary else None,
+            "pending": await newer_than(summary_at) if has_summary else articles,
+        },
+        "factcheck": {
+            "exists": facts > 0,
+            "facts": facts,
+            "last": fc_at,
+            "pending": await newer_than(fc_at) if facts else articles,
+        },
+        "geoparse": {
+            "exists": geo_at is not None,
+            "last": geo_at,
+            "pending": geo_pending,
+        },
+        "snapshots": await one(
+            "SELECT COUNT(*) FROM incident_snapshots WHERE incident_id = ?", (incident_id,)
+        ) or 0,
+        "events": await one(
+            "SELECT COUNT(*) FROM incident_events WHERE incident_id = ?", (incident_id,)
+        ) or 0,
+    }
+
+
+@router.get("/{incident_id}/run-status")
+async def run_stage_status(
+    incident_id: int,
+    current_user: dict = Depends(get_current_user),
+    db: aiosqlite.Connection = Depends(db_dependency),
+):
+    """Status des zuletzt/aktuell laufenden Bausteins."""
+    from agents import stage_runners
+    await _check_incident_access(db, incident_id, current_user["id"], current_user.get("tenant_id"))
+    return {"state": stage_runners.get_state(incident_id)}
+
+
+@router.get("/{incident_id}/factcheck-runs")
+async def list_factcheck_runs(
+    incident_id: int,
+    current_user: dict = Depends(get_current_user),
+    db: aiosqlite.Connection = Depends(db_dependency),
+):
+    """Faktencheck-Historie (archivierte Laeufe mit Zeitstempel)."""
+    await _check_incident_access(db, incident_id, current_user["id"], current_user.get("tenant_id"))
+    cur = await db.execute(
+        "SELECT id, created_at, fact_count FROM fact_check_runs WHERE incident_id = ? ORDER BY created_at DESC, id DESC LIMIT 50",
+        (incident_id,),
+    )
+    return {"runs": [dict(r) for r in await cur.fetchall()]}
+
+
+@router.get("/{incident_id}/factcheck-runs/{run_id}")
+async def factcheck_run_detail(
+    incident_id: int,
+    run_id: int,
+    current_user: dict = Depends(get_current_user),
+    db: aiosqlite.Connection = Depends(db_dependency),
+):
+    """Faktenstand eines archivierten Laufs."""
+    await _check_incident_access(db, incident_id, current_user["id"], current_user.get("tenant_id"))
+    cur = await db.execute(
+        "SELECT facts_json, created_at FROM fact_check_runs WHERE id = ? AND incident_id = ?",
+        (run_id, incident_id),
+    )
+    row = await cur.fetchone()
+    if not row:
+        raise HTTPException(status_code=404, detail="Lauf nicht gefunden.")
+    try:
+        facts = json.loads(row["facts_json"] or "[]")
+    except (ValueError, TypeError):
+        facts = []
+    return {"created_at": row["created_at"], "facts": facts}
+
+
 
 def _slugify(text: str) -> str:
     """Dateinamen-sicherer Slug aus Titel."""