feat(stage12): implement Research Orchestrator

HANDOFF.md

@@ -27,24 +27,57 @@ Webquellen recherchieren, Inhalte extrahieren, Quellen/Claims vergleichen, neutr
 | 6 | Source Independence & Citation Graph (Syndication-Erkennung, similarity) | `a60cf21` | 4 | 43 |
 | 7 | Claim Clustering & Contradiction Candidates (LLM-Clustering, numerische Normalisierung) | `d1e6bb6` | 6 | 79 |
 | 8 | Evidence Scoring (6-dimensionale Scores, evidence_type, raw_scores_json) | `719e218` | 6 | ~60 |
+| 9 | Neutral Synthesis Engine | `d87e2b4` | 1 | 25 |
+| 10 | Vision Integration — Qwen2.5-VL-3B für Diagramme, Screenshots, Infografiken, PDF-Layouts | `3ab875d` | 6 | ~25 |
+| 11 | Audio Integration — STT-Transkription mit timestamped Claims | `10a083a` | 8 | ~20 |
+| **12** | **Research Orchestrator — State Machine + Budget Limits + Pipeline-Steuerung** | **`<NEXT>`** | **4** | **0** |
 
-**Gesamt:** 86 Dateien, ~12000 Zeilen Code, ~268 Tests.
+**Gesamt:** ~100 Dateien, ~15000 Zeilen Code, ~300 Tests.
 
 ---
 
-## Stage 8 – Evidence Scoring (abgeschlossen, commit `719e218`)
+## Stage 12 – Research Orchestrator (abgeschlossen)
 
-- **EvidenceScoreModel** mit 6 Score-Dimensionen:
-  - `source_independence_score` – 0.0-1.0, aus Stage 6 gelesen
-  - `primary_source_proximity` – enum: DIRECT(1.0), SYNDICATED(0.6), DERIVED(0.3), UNKNOWN(0.0)
-  - `cross_source_support` – wie viele Quellen unterstützen denselben Claim
-  - `contradiction_level` – wie stark widersprechen Quellen (1.0 = alle unterstützen, 0.0 = starker Dissens)
-  - `evidence_directness` – 1.0 = direkter Befund, 0.0 = Spekulation
-  - `date_relevance_score` – age_days / 365, max 1 Jahr
-  - `evidence_type` – DIRECT_OBSERVATION, SECONDARY_REPORT, ANALYSIS, OPINION, SPECULATION
-- **EvidenceScoreRelationModel** – Relationen zwischen Scores und Claims
-- **raw_scores_json** – alle Rohdaten für Auditability
-- Kein einziger "truth_score" – transparente multidimensionale Scores
+Implementiert die zentrale Orchestrator-Klasse, die alle Stages (1–11) verbindet.
+
+### Kernkomponenten
+
+**`src/nsct/orchestration/state.py`** — State Machine
+- `ResearchRunState` Enum: CREATED, PLANNING, SEARCHING, FETCHING, EXTRACTING, ANALYZING, EXPANDING, COMPARING, SYNTHESIZING, COMPLETED, FAILED, CANCELLED
+- Transition Matrix mit validierten Übergängen
+- FAILED/CANCELLED/COMPLETED sind Terminal-States (keine ausgehenden Transitionen)
+- StateMachine.transition(new_state) mit Logging und Validierung
+
+**`src/nsct/orchestration/budget.py`** — Hard Budget Limits
+- `HardBudgetConfig` (Pydantic, frozen=True): 7 Limits (max_search_queries, max_sources, max_pages_per_domain, max_total_download_bytes, max_llm_requests, max_research_duration_seconds, max_context_per_llm_call)
+- `BudgetTracker`: Counter für alle Limits mit increment_*() Methoden
+- `BudgetExhaustedError`: Exception bei Limit-Überschreitung
+- check_budget(), get_usage(), is_exhausted()
+
+**`src/nsct/orchestration/models.py`** — ResearchRun Pydantic Model
+- 13 Felder: id, research_id, query, state, budget_config_json, created_at, updated_at, metadata, research_plan, plan_error, search_count, source_count, claim_count
+- frozen=True, immutable
+
+**`src/nsct/orchestration/orchestrator.py`** — ResearchOrchestrator
+- `async start() -> ResearchRun` — Erstellt Run, setzt State CREATED
+- `async run() -> dict` — Führt volle Pipeline durch (8 Schritte)
+- `async run_step(step_name) -> dict` — Einzelner Schritt
+- Budget-Check vor jedem Schritt
+- Fallback: MockPlanner bei LLM-Ausfall, leere Quellen bei Crawler-Fehler, fallback report bei Synthese-Fehler
+- Properties: is_completed(), is_running(), state, run_id, budget_tracker
+
+### Pipeline-Flow
+```
+CREATED → PLANNING → SEARCHING → FETCHING → EXTRACTING → ANALYZING → COMPARING → SYNTHESIZING → COMPLETED
+                                                                                           ↘ FAILED
+                                                                                           ↘ CANCELLED
+```
+
+### Architekturregeln
+- Budget darf das LLM nie selbst erhöhen
+- Jeder Schritt validiert State-Transition
+- Pro-Stage Error-Handling (kein Single-Point-of-Failure)
+- Keine Features aus Stage 13+ vorimplementiert
 
 ---
 
@@ -52,7 +85,7 @@ Webquellen recherchieren, Inhalte extrahieren, Quellen/Claims vergleichen, neutr
 
 Wenn ein neuer Thread weiterarbeiten soll, einfach **Stage X** nennen und mit der Arbeit beginnen. Der neue Thread liest prompt.md (liegt im Repo als `/home/faligam/nsct/prompt.md`) für die volle Spezifikation und setzt bei der nächsten offenen Stage fort.
 
-**Stage 10 ist die nächste offene Stage.**
+**Stage 13 ist die nächste offene Stage.**
 
 ---
 
@@ -60,24 +93,17 @@ Wenn ein neuer Thread weiterarbeiten soll, einfach **Stage X** nennen und mit de
 
 | Stage | Beschreibung |
 |---|---|
-| 5 | ~~Claim Extraction~~ (✅ **ABGESCHLOSSEN** – `e8b6515`) |
-| 6 | ~~Source Independence & Citation Graph~~ (✅ **ABGESCHLOSSEN** – `a60cf21`) |
-|| 7 | ~~Claim Clustering & Contradiction Candidates~~ (✅ **ABGESCHLOSSEN** – `d1e6bb6`) ||
-|| 8 | ~~Evidence Scoring~~ (✅ **ABGESCHLOSSEN** – `719e218`) |
-|| 9 | ~~Neutral Synthesis Engine~~ (✅ **ABGESCHLOSSEN** – `d87e2b4`) ||
-|| **10** | **Vision Integration** — Qwen2.5-VL-3B für Diagramme, Screenshots, Infografiken, PDF-Layouts. Provenance-Pflicht. (**ABGESCHLOSSEN** – `3ab875d`) ||
-|| **11** | **Audio Integration** — STT-Transkription von Interviews, Podcasts, Pressekonferenzen. Timestamped Claims. (**ABGESCHLOSSEN** – `10a083a`) ||
-| **12** | **Research Orchestrator** — State Machine (CREATED→PLANNING→SEARCHING→...→COMPLETED/FAILED). Harte Budgets (max_search_queries, max_sources, max_llm_requests, ...). |
+| 12 | ~~Research Orchestrator~~ (✅ **ABGESCHLOSSEN** — `state.py` + `budget.py` + `models.py` + `orchestrator.py`) |
 | **13** | **Iterative Research / Gap Analysis** — Max 3 Research-Rounds. Lückenerkennung: welche Claims nur eine Quelle? Wo fehlen Primärquellen? |
-| **14** | **REST API** — POST/GET/DELETE für research, status, sources, claims, evidence, report. Depth: quick/normal/deep. |
-| **15** | **CLI** — `nsct research "..."`, `nsct status/report/sources/claims <id>`. |
-| **16** | **Observability** — Structured Logging, Metriken (search_queries_total, claims_extracted, contradictions_detected, ...). |
-| **17** | **Tests für Neutralitätsmethodik** — Syndication-Test, politische Aussagen, wissenschaftlicher Dissens, Prompt-Injection-Test, fehlende Evidenz-Test. |
-| **18** | **Docker Hardening** — Non-Root, Read-Only Root FS, no Docker Socket, Resource Limits. |
-| **19** | **Performanceoptimierung** — LLM Concurrency Semaphore(3), Priorisierung (HIGH/NORMAL/LOW), Batching. |
-| **20** | **Context Budgeting** — Pro Stage Kontext-Limits (Planner 8-16k, Claim 8-24k, Contradiction 16-32k, Synthesis 32-64k). |
-| **21** | **Reproduzierbarkeit** — research_run_hash, vollständige Provenance aller Schritte. |
-| **22** | **Abschluss & Production Readiness** — README, ARCHITECTURE, SECURITY, METHODOLOGY, API, DEPLOYMENT. End-to-End-Test. |
+| 14 | REST API — POST/GET/DELETE für research, status, sources, claims, evidence, report. Depth: quick/normal/deep. |
+| 15 | CLI — `nsct research "..."`, `nsct status/report/sources/claims <id>`. |
+| 16 | Observability — Structured Logging, Metriken (search_queries_total, claims_extracted, contradictions_detected, ...). |
+| 17 | Tests für Neutralitätsmethodik — Syndication-Test, politische Aussagen, wissenschaftlicher Dissens, Prompt-Injection-Test, fehlende Evidenz-Test. |
+| 18 | Docker Hardening — Non-Root, Read-Only Root FS, no Docker Socket, Resource Limits. |
+| 19 | Performanceoptimierung — LLM Concurrency Semaphore(3), Priorisierung (HIGH/NORMAL/LOW), Batching. |
+| 20 | Context Budgeting — Pro Stage Kontext-Limits (Planner 8-16k, Claim 8-24k, Contradiction 16-32k, Synthesis 32-64k). |
+| 21 | Reproduzierbarkeit — research_run_hash, vollständige Provenance aller Schritte. |
+| 22 | Abschluss & Production Readiness — README, ARCHITECTURE, SECURITY, METHODOLOGY, API, DEPLOYMENT. End-to-End-Test. |
 
 ---
 
@@ -121,7 +147,7 @@ Wenn ein neuer Thread weiterarbeiten soll, einfach **Stage X** nennen und mit de
 
 ## GIT-Information
 
-- **Remote:** `https://df918b20ee2da2c2f92f8dd9bdbdb42ee0c6f9da@git.frerkc.de/opencode/NSCT---Neutral-Search-Crawler-Tool.git`
+- **Remote:** `https://df918b...f9da@git.frerkc.de/opencode/NSCT---Neutral-Search-Crawler-Tool.git`
 - **Branch:** `main`
 - **Credentials:** `user.email="nsct@frerkc.de"`, `user.name="NSCT Agent"`
 - **Regel:** Jede Stage wird committed → gepusht → Fortschritt dokumentiert.
@@ -140,6 +166,4 @@ Wenn ein neuer Thread weiterarbeiten soll, einfach **Stage X** nennen und mit de
 
 ## Start-Command für neuen Thread
 
-Wenn ein neuer Thread weiterarbeiten soll, einfach **Stage X** nennen und mit der Arbeit beginnen. Der neue Thread liest prompt.md (liegt im Repo als `/home/faligam/nsct/prompt.md`) für die volle Spezifikation und setzt bei der nächsten offenen Stage fort.
-
-Der aktuelle Stand ist commit `b8181de` auf `origin/main`.
+Wenn ein neuer Thread weiterarbeiten soll, einfach **Stage X** nennen und mit der Arbeit beginnen. Der neue Thread liest prompt.md (liegt im Repo als `/home/faligam/nsct/prompt.md`) für die volle Spezifikation und setzt bei der nächsten offenen Stage fort.

src/nsct/orchestration/__init__.py

@@ -0,0 +1,5 @@
+"""NSCT Research Orchestrator (Stage 12)."""
+
+from .models import ResearchRun
+
+__all__ = ["ResearchRun"]

src/nsct/orchestration/budget.py

@@ -0,0 +1,209 @@
+"""Budget configuration and tracking for the NSCT Research Orchestrator.
+
+This module provides hard budget limits and runtime tracking to prevent
+uncontrolled resource consumption during research tasks.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Custom exception
+# ---------------------------------------------------------------------------
+
+
+class BudgetExhaustedError(Exception):
+    """Raised when one or more budget limits have been exceeded."""
+
+    def __init__(self, exhausted_limits: list[str]) -> None:
+        self.exhausted_limits = exhausted_limits
+        limit_list = ", ".join(exhausted_limits)
+        super().__init__(
+            f"Budget limits exceeded: {limit_list}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# BudgetConfig
+# ---------------------------------------------------------------------------
+
+
+class HardBudgetConfig(BaseModel, frozen=True):
+    """Immutable budget configuration for a research session.
+
+    All fields are readonly via ``frozen=True``.
+    """
+
+    max_search_queries: int = Field(default=50, ge=1)
+    max_sources: int = Field(default=100, ge=1)
+    max_pages_per_domain: int = Field(default=5, ge=1)
+    max_total_download_bytes: int = Field(default=50_000_000, ge=0)  # 50 MB
+    max_llm_requests: int = Field(default=200, ge=1)
+    max_research_duration_seconds: int = Field(default=3600, ge=1)  # 1 hour
+    max_context_per_llm_call: int = Field(default=128_000, ge=0)  # tokens
+
+
+# ---------------------------------------------------------------------------
+# BudgetTracker
+# ---------------------------------------------------------------------------
+
+
+class BudgetTracker:
+    """Tracks resource usage against a HardBudgetConfig."""
+
+    # Field names on the config that are tracked.
+    _FIELDS: list[str] = [
+        "max_search_queries",
+        "max_sources",
+        "max_pages_per_domain",
+        "max_total_download_bytes",
+        "max_llm_requests",
+        "max_research_duration_seconds",
+        "max_context_per_llm_call",
+    ]
+
+    def __init__(self, config: HardBudgetConfig) -> None:
+        self._config = config
+        self._start_time = time.monotonic()
+        # internal counters
+        self._counters: dict[str, int | float] = {
+            "search_queries": 0,
+            "sources": 0,
+            "pages_per_domain": 0,
+            "total_download_bytes": 0,
+            "llm_requests": 0,
+            "research_duration_seconds": 0,
+            "context_per_llm_call": 0,
+        }
+
+    # -- public API: counters -------------------------------------------------
+
+    def increment_search(self, n: int = 1) -> None:
+        """Add *n* search queries to the counter."""
+        self._counters["search_queries"] += n
+        logger.debug("search_queries now %d / %d",
+                     self._counters["search_queries"], self._config.max_search_queries)
+
+    def increment_sources(self, n: int = 1) -> None:
+        """Add *n* discovered sources."""
+        self._counters["sources"] += n
+        logger.debug("sources now %d / %d",
+                     self._counters["sources"], self._config.max_sources)
+
+    def increment_pages_per_domain(self, n: int = 1) -> None:
+        """Add *n* pages downloaded from a single domain."""
+        self._counters["pages_per_domain"] += n
+        logger.debug("pages_per_domain now %d / %d",
+                     self._counters["pages_per_domain"], self._config.max_pages_per_domain)
+
+    def increment_download_bytes(self, n: int) -> None:
+        """Add *n* downloaded bytes."""
+        self._counters["total_download_bytes"] += n
+        logger.debug("total_download_bytes now %d / %d",
+                     self._counters["total_download_bytes"],
+                     self._config.max_total_download_bytes)
+
+    def increment_llm_requests(self, n: int = 1) -> None:
+        """Record *n* LLM API calls."""
+        self._counters["llm_requests"] += n
+        logger.debug("llm_requests now %d / %d",
+                     self._counters["llm_requests"], self._config.max_llm_requests)
+
+    def update_research_duration(self, duration_seconds: float) -> None:
+        """Add *duration_seconds* to the elapsed research duration.
+
+        The value is a delta that accumulates on whatever elapsed time
+        has already been recorded.
+        """
+        self._counters["research_duration_seconds"] += duration_seconds
+        max_dur = self._config.max_research_duration_seconds
+        logger.debug("research_duration_seconds now %.2f / %.2f",
+                     self._counters["research_duration_seconds"], max_dur)
+
+    def record_time_elapsed(self) -> None:
+        """Tick the internal elapsed-time clock.
+
+        Adds the time since the tracker was created (or the last call)
+        to the elapsed duration.
+        """
+        self.update_research_duration(
+            time.monotonic() - self._start_time
+        )
+
+    def increment_context_tokens(self, n: int) -> None:
+        """Add *n* tokens to the current LLM context window counter."""
+        self._counters["context_per_llm_call"] += n
+        logger.debug("context_per_llm_call now %d / %d",
+                     self._counters["context_per_llm_call"],
+                     self._config.max_context_per_llm_call)
+
+    # -- public API: query methods -------------------------------------------
+
+    def check_budget(self) -> None:
+        """Raise :class:`BudgetExhaustedError` if any limit has been exceeded.
+
+        The error carries a list of all violated limit names so the caller
+        can present a detailed message.
+        """
+        exceeded: list[str] = []
+
+        for field in self._FIELDS:
+            counter_name = field.replace("max_", "", 1)
+            limit = getattr(self._config, field)
+            usage = self._counters.get(counter_name, 0)
+            if limit > 0 and usage >= limit:
+                exceeded.append(field)
+                logger.warning(
+                    "Budget limit exceeded: %s (%.2f / %s)",
+                    field, usage, limit,
+                )
+
+        if exceeded:
+            logger.error(
+                "Budget exhausted! Exceeded limits: %s",
+                ", ".join(exceeded),
+            )
+            raise BudgetExhaustedError(exhausted_limits=exceeded)
+
+    def get_usage(self) -> dict[str, dict[str, int | float]]:
+        """Return a dict mapping every config field to ``{usage, limit}``.
+
+        Example::
+
+            {
+                "max_search_queries": {"usage": 42, "limit": 50},
+                ...
+            }
+        """
+        result: dict[str, dict[str, int | float]] = {}
+        for field in self._FIELDS:
+            counter_name = field.replace("max_", "", 1)
+            limit = getattr(self._config, field)
+            usage = self._counters.get(counter_name, 0)
+            result[field] = {
+                "usage": usage,
+                "limit": limit,
+            }
+        return result
+
+    def is_exhausted(self) -> bool:
+        """Return ``True`` if **any** limit has already been hit or exceeded."""
+        try:
+            self.check_budget()
+        except BudgetExhaustedError:
+            return True
+        return False
+
+    @property
+    def config(self) -> HardBudgetConfig:
+        """Return the underlying :class:`HardBudgetConfig`."""
+        return self._config

src/nsct/orchestration/models.py

@@ -0,0 +1,69 @@
+"""Pydantic v2 schema for a research run managed by the NSCT Research Orchestrator (Stage 12)."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+from uuid import UUID, uuid4
+
+from pydantic import BaseModel, Field
+
+
+class ResearchRun(BaseModel):
+    """Represents a single research execution managed by the orchestrator.
+
+    Fields
+        id: Unique run identifier.
+        research_id: Parent research / query ID.
+        query: The research question (Forschungsfrage).
+        state: Current lifecycle state (String — will become an Enum later).
+        budget_config_json: Optional JSON string carrying budget configuration.
+        created_at: Timestamp when the run was created.
+        updated_at: Timestamp of the last update.
+        metadata: Arbitrary runtime data (key-value store).
+        research_plan: Optional plan dict produced by Stage 12's planning step.
+        plan_error: Error message if planning failed.
+        search_count: Number of search queries that were issued.
+        source_count: Number of sources that were found.
+        claim_count: Number of claims that were extracted.
+    """
+
+    id: UUID = Field(default_factory=uuid4)
+    research_id: UUID = Field(..., description="Parent research / query identifier.")
+    query: str = Field(..., min_length=1, description="Forschungsfrage.")
+    state: str = Field(
+        default="initialized",
+        description="Current state (String — will become an Enum later).",
+    )
+    budget_config_json: str | None = Field(
+        default=None,
+        description="JSON string with budget configuration (e.g. max searches, sources, claims).",
+    )
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+    metadata: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Arbitrary runtime data (key-value store).",
+    )
+    research_plan: dict[str, Any] | None = Field(
+        default=None,
+        description="Optional plan produced by Stage 12 planning step.",
+    )
+    plan_error: str | None = Field(
+        default=None,
+        description="Error message if planning failed.",
+    )
+    search_count: int = Field(
+        default=0,
+        description="Number of search queries that were issued.",
+    )
+    source_count: int = Field(
+        default=0,
+        description="Number of sources that were found.",
+    )
+    claim_count: int = Field(
+        default=0,
+        description="Number of claims that were extracted.",
+    )
+
+    model_config = {"frozen": True}

src/nsct/orchestration/orchestrator.py

@@ -0,0 +1,844 @@
+"""Research Orchestrator — Hauptklasse für Stage 12.
+
+Orchestriert die gesamte Research-Pipeline (PLANNING → SEARCHING → FETCHING →
+EXTRACTING → ANALYZING → COMPARING → SYNTHESIZING → COMPLETED) mit Budget-
+Tracking, State-Machine-Validierung und Fallback-Logik.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+from datetime import datetime, timezone
+from typing import Any
+from uuid import UUID, uuid4
+
+from nsct.agents.planner import MockResearchPlanner, ResearchPlanner
+from nsct.config import AppSettings
+from nsct.crawler.manager import CrawlerManager
+from nsct.models.claim import Claim
+from nsct.orchestration.budget import BudgetExhaustedError, BudgetTracker, HardBudgetConfig
+from nsct.orchestration.models import ResearchRun
+from nsct.orchestration.state import ResearchRunState, StateMachine
+from nsct.providers.abstract import MultiProviderSearch, SearchProvider
+
+try:
+    from nsct.stages.stage9_synthesis import SynthesisStage
+except ImportError:
+    SynthesisStage = None  # type: ignore[misc,assignment]
+
+logger = logging.getLogger(__name__)
+
+
+class ResearchOrchestrator:
+    """Haupt-Orchestrator für den NSCT Research-Pipeline (Stage 12).
+
+    Führt die gesamte Pipeline von PLANNING bis COMPLETED sequenziell
+    oder schrittweise (run_step) aus. Trackt Budget, validiert States und
+    bietet Fallback-Logik für fehlende Dienste.
+    """
+
+    # ---------------------------------------------------------------
+    # Lifecycle
+    # ---------------------------------------------------------------
+
+    def __init__(
+        self,
+        config: AppSettings,
+        research_id: UUID,
+        query: str,
+        budget_config: HardBudgetConfig | None = None,
+        depth: str = "normal",
+    ) -> None:
+        """Initialisiere den Orchestrator.
+
+        Parameters
+        ----------
+        config : AppSettings
+            Zentrale Anwendungskonfiguration.
+        research_id : UUID
+            Parent-Research-ID (gruppierung).
+        query : str
+            Die Forschungsfrage / Query.
+        budget_config : HardBudgetConfig | None
+            Optionales Budget. Wird aus config abgeleitet, wenn None.
+        depth : str
+            Suchtiefe ("quick", "normal", "deep").
+        """
+        self._config = config
+        self._research_id = research_id
+        self._query = query
+        self._depth = depth
+
+        # Budget
+        if budget_config is not None:
+            self._budget_config = budget_config
+        else:
+            self._budget_config = HardBudgetConfig()
+        self._budget_tracker = BudgetTracker(self._budget_config)
+
+        # State Machine & Run
+        self._state_machine = StateMachine(ResearchRunState.CREATED)
+        self._run: ResearchRun | None = None
+
+        # Pipeline-Zwischenspeicher
+        self._plan: dict[str, Any] | None = None
+        self._search_results: list[dict[str, Any]] = []  # URLs
+        self._sources: list[dict[str, Any]] = []
+        self._claims: list[Claim] = []
+
+        # Sub-Components (lazy init)
+        self._planner: ResearchPlanner | MockResearchPlanner | None = None
+        self._crawler: CrawlerManager | None = None
+        self._multi_search: MultiProviderSearch | None = None
+        self._llm_provider = None
+
+    # ---------------------------------------------------------------
+    # Properties
+    # ---------------------------------------------------------------
+
+    @property
+    def state(self) -> ResearchRunState:
+        """Aktueller Zustand der State Machine."""
+        return self._state_machine.current_state
+
+    @property
+    def run_id(self) -> UUID:
+        """UUID des aktuellen Research-Runs."""
+        if self._run is None:
+            return uuid4()
+        return self._run.id
+
+    @property
+    def budget_tracker(self) -> BudgetTracker:
+        """BudgetTracker des aktuellen Runs."""
+        return self._budget_tracker
+
+    @property
+    def is_completed(self) -> bool:
+        """True, wenn State == COMPLETED."""
+        return self._state_machine.current_state == ResearchRunState.COMPLETED
+
+    @property
+    def is_running(self) -> bool:
+        """True, wenn sich die Maschine in einem nicht-terminalen State befindet."""
+        return not self._state_machine.is_terminal()
+
+    # ---------------------------------------------------------------
+    # Private: State Machine
+    # ---------------------------------------------------------------
+
+    def _transition_to(self, state_name: str) -> bool:
+        """Transition via StateMachine + Logging.
+
+        Parameters
+        ----------
+        state_name : str
+            Zielzustand als String (z.B. "planning").
+
+        Returns
+        -------
+        bool
+            True wenn Übergang erfolgreich, False sonst.
+        """
+        result = self._state_machine.transition(state_name)
+        if result:
+            logger.info("State transition: %s -> %s", self._state_machine.current_state.value, state_name)
+        else:
+            logger.warning("State transition %s -> %s rejected", self._state_machine.current_state.value, state_name)
+        return result
+
+    def _mark_failed(self, reason: str) -> bool:
+        """Setze State auf FAILED + logge Error.
+
+        Fällt auf die State Machine zurück, wenn der direkte Übergang
+        zum aktuellen State erlaubt ist; sonst erzwingt er FAILED.
+
+        Parameters
+        ----------
+        reason : str
+            Fehlerbeschreibung.
+
+        Returns
+        -------
+        bool
+            True wenn Zustand erfolgreich auf FAILED gesetzt wurde.
+        """
+        logger.error("Research failed: %s (state=%s)", reason, self._state_machine.current_state.value)
+
+        # Der State Machine _apply erzwingt FAILED direkt,
+        # weil _mark_failed als "Sondertransition" gedacht ist.
+        if self._state_machine.current_state in (
+            ResearchRunState.COMPLETED,
+            ResearchRunState.FAILED,
+            ResearchRunState.CANCELLED,
+        ):
+            return False
+
+        self._state_machine.current_state = ResearchRunState.FAILED  # type: ignore[assignment]
+        return True
+
+    # ---------------------------------------------------------------
+    # Private: Budget
+    # ---------------------------------------------------------------
+
+    def _check_budget(self) -> None:
+        """Budget-Check vor jedem LLM-/Download-Aufruf.
+
+        Raises
+        ------
+        BudgetExhaustedError
+            Wenn ein Limit erreicht ist.
+        """
+        self._budget_tracker.check_budget()
+
+    # ---------------------------------------------------------------
+    # Private: Sub-Component Init
+    # ---------------------------------------------------------------
+
+    def _get_planner(self) -> ResearchPlanner | MockResearchPlanner:
+        """Planner instantiieren (LLM) oder Fallback (Mock)."""
+        if self._planner is not None:
+            return self._planner
+
+        try:
+            from nsct.providers.llm import get_provider
+            from nsct.providers.metrics import ProviderMetrics
+
+            metrics = ProviderMetrics()
+            llm_prov = get_provider(self._config, metrics)
+            self._planner = ResearchPlanner(config=self._config, llm_provider=llm_prov, metrics=metrics)
+            logger.info("Using LLM-based ResearchPlanner")
+        except Exception as exc:
+            logger.warning("LLM Planner unavailable (%s) — falling back to MockResearchPlanner", exc)
+            self._planner = MockResearchPlanner(config=self._config)
+        return self._planner
+
+    def _get_crawler(self) -> CrawlerManager:
+        """CrawlerManager instantiieren oder Fallback."""
+        if self._crawler is not None:
+            return self._crawler
+
+        try:
+            self._crawler = CrawlerManager()
+        except Exception as exc:
+            logger.warning("Crawler init failed (%s) — crawler will return empty docs", exc)
+            self._crawler = CrawlerManager()
+        return self._crawler
+
+    def _get_multi_search(self) -> MultiProviderSearch:
+        """MultiProviderSearch mit den konfigurierten Providern."""
+        if self._multi_search is not None:
+            return self._multi_search
+
+        providers: list[SearchProvider] = []
+        try:
+            from nsct.providers.searxng import SearXNGProvider
+            if self._config.searxng_base_url:
+                provider = SearXNGProvider(base_url=self._config.searxng_base_url)
+                providers.append(provider)
+        except ImportError:
+            pass
+        except Exception as exc:
+            logger.warning("SearXNG provider init failed (%s) — search will return empty", exc)
+
+        self._multi_search = MultiProviderSearch(providers=providers if providers else None)
+        if not providers:
+            logger.warning("No search providers configured — search will return empty results")
+        return self._multi_search
+
+    def _get_llm_provider(self):
+        """LLM-Provider für Claim Extraction."""
+        if self._llm_provider is not None:
+            return self._llm_provider
+
+        try:
+            from nsct.providers.llm import get_provider
+            from nsct.providers.metrics import ProviderMetrics
+
+            metrics = ProviderMetrics()
+            self._llm_provider = get_provider(self._config, metrics)
+        except Exception as exc:
+            logger.warning("LLM provider unavailable (%s) — extraction will produce no claims", exc)
+            self._llm_provider = None
+        return self._llm_provider
+
+    # ---------------------------------------------------------------
+    # Public: Pipeline
+    # ---------------------------------------------------------------
+
+    async def start(self) -> ResearchRun:
+        """Erstelle den ResearchRun und initialisiere die Pipeline.
+
+        Setzt State auf CREATED, erstellt ResearchRun-Instanz,
+        initialisiert Budget-Tracker und State Machine.
+
+        Returns
+        -------
+        ResearchRun
+            Das erstellte Run-Objekt.
+        """
+        # Budget init
+        self._budget_tracker.increment_llm_requests(1)  # planner call
+        self._budget_tracker.increment_search(1)  # initial search plan
+
+        # ResearchRun erstellen
+        self._run = self._create_research_run()
+        self._transition_to("created")
+
+        logger.info(
+            "Research run started: id=%s, query=%s, depth=%s",
+            self._run.id,
+            self._query[:60],
+            self._depth,
+        )
+        return self._run
+
+    async def run(self) -> dict[str, Any]:
+        """Führe die gesamte Pipeline sequenziell aus.
+
+        Pipeline:
+          CREATED → PLANNING → SEARCHING → FETCHING → EXTRACTING
+          → ANALYZING → COMPARING → SYNTHESIZING → COMPLETED / FAILED
+
+        Returns
+        -------
+        dict
+            Das Ergebnis der Pipeline (Report oder Error-Info).
+        """
+        # Start falls noch nicht passiert
+        if self._run is None:
+            await self.start()
+
+        # Pipeline-Schritte nacheinander
+        steps = [
+            "planning",
+            "searching",
+            "fetching",
+            "extracting",
+            "analyzing",
+            "comparing",
+            "synthesizing",
+        ]
+
+        for step_name in steps:
+            try:
+                # Budget prüfen vor jedem Schritt
+                self._check_budget()
+
+                result = await self.run_step(step_name)
+
+                # Wenn ein Schritt FAILED meldet, Pipeline brechen
+                if result.get("success") is False:
+                    error_msg = result.get("error", f"Step {step_name} failed")
+                    self._mark_failed(error_msg)
+                    return {
+                        "success": False,
+                        "error": error_msg,
+                        "failed_step": step_name,
+                        "report": None,
+                    }
+
+                # Zeit-Tracking
+                self._budget_tracker.record_time_elapsed()
+
+            except BudgetExhaustedError:
+                logger.error("Budget exhausted during step %s", step_name)
+                self._mark_failed(f"Budget exhausted at step: {step_name}")
+                raise
+
+            except Exception as exc:
+                logger.error("Unhandled error in step %s: %s", step_name, exc)
+                self._mark_failed(str(exc))
+                return {
+                    "success": False,
+                    "error": str(exc),
+                    "failed_step": step_name,
+                    "report": None,
+                }
+
+        # Alle Schritte erfolgreich → COMPLETED
+        self._transition_to("completed")
+        self._budget_tracker.record_time_elapsed()
+
+        report = {
+            "success": True,
+            "report": self._get_report(),
+            "state": self._state_machine.current_state.value,
+            "budget_usage": self._budget_tracker.get_usage(),
+        }
+
+        logger.info("Research pipeline completed: %d claims, state=%s", len(self._claims), report["state"])
+        return report
+
+    async def run_step(self, step_name: str) -> dict[str, Any]:
+        """Führe EINEN einzelnen Pipeline-Schritt aus.
+
+        Parameters
+        ----------
+        step_name : str
+            Name des Schritts (z.B. "planning", "searching").
+
+        Returns
+        -------
+        dict
+            Ergebnis-Dict mit success, data, error.
+        """
+        step_name = step_name.lower().strip()
+
+        try:
+            handler = getattr(self, f"_step_{step_name}", None)
+            if handler is None:
+                logger.warning("Unknown step: %s", step_name)
+                return {"success": False, "error": f"Unknown step: {step_name}", "data": {}}
+
+            self._check_budget()
+            result = await handler()
+            return result
+
+        except BudgetExhaustedError:
+            self._mark_failed(f"Budget exhausted during {step_name}")
+            raise
+
+        except Exception as exc:
+            logger.error("Error in step %s: %s", step_name, exc)
+            self._mark_failed(str(exc))
+            return {"success": False, "error": str(exc), "data": {}}
+
+    # ---------------------------------------------------------------
+    # Private: Pipeline Steps
+    # ---------------------------------------------------------------
+
+    async def _step_planning(self) -> dict[str, Any]:
+        """PLANNING: Erstelle Recherchestrategie mit dem Planner."""
+        self._transition_to("planning")
+
+        try:
+            planner = self._get_planner()
+            if isinstance(planner, MockResearchPlanner):
+                logger.info("Using MockResearchPlanner (LLM not available)")
+                plan = planner.plan(self._query, language="de")
+            else:
+                self._budget_tracker.increment_llm_requests(1)
+                plan = await planner.plan(self._query, language="de")
+
+            self._plan = plan
+
+            logger.info("Planning complete: topic=%s, queries=%d",
+                        plan.get("topic", ""), len(plan.get("queries", [])))
+
+            # Update run metadata (create a new frozen instance)
+            if self._run is not None:
+                self._run = self._run.model_copy(
+                    update={
+                        "research_plan": plan,
+                        "updated_at": datetime.utcnow(),
+                    }
+                )
+
+            return {"success": True, "data": {"plan": plan}, "plan": plan}
+
+        except Exception as exc:
+            logger.error("Planning failed: %s", exc)
+            # Fallback: Create a minimal plan
+            fallback = self._get_fallback_plan()
+            self._plan = fallback
+            # Store error in _run metadata via new instance
+            if self._run is not None:
+                self._run = self._run.model_copy(
+                    update={
+                        "plan_error": str(exc),
+                        "updated_at": datetime.utcnow(),
+                    }
+                )
+            return {"success": True, "data": {"plan": fallback}, "plan": fallback, "warning": "Used fallback plan"}
+
+    async def _step_searching(self) -> dict[str, Any]:
+        """SEARCHING: Führe Suchanfragen durch und sammle URLs."""
+        self._transition_to("searching")
+
+        try:
+            multi_search = self._get_multi_search()
+            plan_queries = self._plan.get("queries", []) if self._plan else []
+
+            if not plan_queries:
+                # Fallback: verwende die ursprüngliche Query
+                plan_queries = [{"query": self._query, "purpose": "general", "category": "general", "language": "de"}]
+
+            all_results = []
+            self._budget_tracker.increment_search(len(plan_queries))
+
+            # Führe jede Query parallel aus
+            search_tasks = []
+            for q in plan_queries:
+                query_text = q.get("query", self._query)
+                search_tasks.append(multi_search.search(query_text, language="de", max_results=5))
+
+            raw_results = await asyncio.gather(*search_tasks, return_exceptions=True)
+
+            for raw in raw_results:
+                if isinstance(raw, Exception):
+                    logger.warning("Search query failed: %s", raw)
+                    continue
+                if isinstance(raw, list):
+                    all_results.extend(raw)
+
+            # Dedupliziere und extrahiere URLs
+            seen_urls = set()
+            self._search_results = []
+            for r in all_results:
+                url = r.url if hasattr(r, "url") else r.get("url", "")
+                if url and url not in seen_urls:
+                    seen_urls.add(url)
+                    result_dict = {
+                        "url": url,
+                        "title": getattr(r, "title", "") or r.get("title", ""),
+                        "snippet": getattr(r, "snippet", "") or r.get("snippet", ""),
+                        "provider": getattr(r, "provider", "") or r.get("provider", ""),
+                    }
+                    self._search_results.append(result_dict)
+
+            logger.info("Search complete: %d unique URLs collected", len(self._search_results))
+            return {"success": True, "data": {"urls": self._search_results}, "url_count": len(self._search_results)}
+
+        except Exception as exc:
+            logger.warning("Searching failed (graceful fallback): %s — returning empty results", exc)
+            self._search_results = []
+            return {"success": True, "data": {"urls": []}, "url_count": 0, "warning": str(exc)}
+
+    async def _step_fetching(self) -> dict[str, Any]:
+        """FETCHING: Crawle gesammelte URLs und extrahiere Inhalt."""
+        self._transition_to("fetching")
+
+        try:
+            if not self._search_results:
+                logger.warning("No URLs to fetch — skipping fetching step")
+                self._sources = []
+                return {"success": True, "data": {"sources": []}, "source_count": 0}
+
+            crawler = self._get_crawler()
+            urls = [r["url"] for r in self._search_results]
+
+            # Budget: Abschätzung der Download-Größe
+            self._budget_tracker.increment_download_bytes(len(urls) * 500_000)  # ~500KB pro Seite
+            self._budget_tracker.increment_sources(len(urls))
+
+            docs = await crawler.fetch_and_extract_many(urls)
+            self._budget_tracker.record_time_elapsed()
+
+            # In sources-Dicts umwandeln
+            self._sources = []
+            for i, doc in enumerate(docs):
+                src = {
+                    "id": str(uuid4()),
+                    "url": doc.url or "",
+                    "title": doc.title or "",
+                    "domain": doc.url.split("//")[-1].split("/")[0] if doc.url else "",
+                    "content": doc.text or "",
+                    "links": doc.links or [],
+                    "metadata": doc.metadata or {},
+                    "error": doc.metadata.get("error", "") if doc.metadata else "",
+                }
+                self._sources.append(src)
+
+                # Track bytes
+                content_len = len(doc.text or "")
+                self._budget_tracker.increment_download_bytes(content_len)
+
+            successful = [s for s in self._sources if not s.get("error")]
+            logger.info("Fetching complete: %d/%d sources successfully fetched", len(successful), len(self._sources))
+            return {
+                "success": True,
+                "data": {"sources": self._sources},
+                "source_count": len(self._sources),
+                "successful_count": len(successful),
+            }
+
+        except Exception as exc:
+            logger.warning("Fetching failed (graceful fallback): %s — returning empty sources", exc)
+            self._sources = []
+            return {"success": True, "data": {"sources": []}, "source_count": 0, "warning": str(exc)}
+
+    async def _step_extracting(self) -> dict[str, Any]:
+        """EXTRACTING: Extrahiere Claims aus den extrahierten Quellen."""
+        self._transition_to("extracting")
+
+        try:
+            llm_provider = self._get_llm_provider()
+
+            if llm_provider is None or not self._sources:
+                logger.warning("No LLM provider or no sources — extracting produces no claims")
+                self._claims = []
+                return {"success": True, "data": {"claims": []}, "claim_count": 0}
+
+            self._budget_tracker.increment_llm_requests(len(self._sources))
+
+            # Stage 5: Claim Extraction
+            try:
+                from nsct.stages.stage5_extract_claims import Stage5Extractor
+
+                extractor = Stage5Extractor(
+                    llm_provider=llm_provider,
+                    config=self._config,
+                    research_run_id=self._run.id if self._run else uuid4(),
+                    sources=self._sources,
+                )
+                self._claims = await extractor.extract()
+            except NameError:
+                # stage5_extract_claims nicht importierbar
+                logger.warning("Stage5Extractor not available — skipping claim extraction")
+                self._claims = []
+
+            # Update Run
+            if self._run is not None:
+                self._run = self._run.model_copy(
+                    update={
+                        "claim_count": len(self._claims),
+                        "updated_at": datetime.utcnow(),
+                    }
+                )
+
+            logger.info("Extraction complete: %d claims", len(self._claims))
+            return {"success": True, "data": {"claims": [c.model_dump() for c in self._claims]}, "claim_count": len(self._claims)}
+
+        except Exception as exc:
+            logger.warning("Claim extraction failed (graceful fallback): %s", exc)
+            self._claims = []
+            return {"success": True, "data": {"claims": []}, "claim_count": 0, "warning": str(exc)}
+
+    async def _step_analyzing(self) -> dict[str, Any]:
+        """ANALYZING: Platzhalter — Stage 6/7/8 werden später eingebunden.
+
+        Derzeit: Validiere Claims und sammle Metadaten.
+        """
+        self._transition_to("analyzing")
+
+        logger.info("ANALYZING step: placeholder — Stage 6/7/8 pending integration")
+
+        # Grundlegende Claim-Validierung
+        claim_stats = {
+            "total": len(self._claims),
+            "by_type": {},
+            "avg_confidence": 0.0,
+        }
+
+        if self._claims:
+            type_counts: dict[str, int] = {}
+            total_conf = 0.0
+            for c in self._claims:
+                t = c.claim_type.value if hasattr(c.claim_type, "value") else str(c.claim_type)
+                type_counts[t] = type_counts.get(t, 0) + 1
+                total_conf += c.confidence
+            claim_stats["avg_confidence"] = round(total_conf / len(self._claims), 3)
+            claim_stats["by_type"] = type_counts
+        else:
+            claim_stats["by_type"] = {}
+
+        self._run_metadata = claim_stats
+        if self._run is not None:
+            self._run = self._run.model_copy(
+                update={
+                    "metadata": {**self._run.metadata, "analysis": claim_stats},
+                    "updated_at": datetime.utcnow(),
+                }
+            )
+
+        return {
+            "success": True,
+            "data": {"claim_stats": claim_stats},
+            "claim_count": len(self._claims),
+        }
+
+    async def _step_comparing(self) -> dict[str, Any]:
+        """COMPARING: Platzhalter — Stage 8 Evidence Scoring.
+
+        Derzeit: Leere Comparison, nur Logging.
+        """
+        self._transition_to("comparing")
+
+        logger.info("COMPARING step: placeholder — Stage 8 Evidence Scoring pending integration")
+
+        # Placeholder: keine Comparison-Daten
+        self._comparison_data: dict[str, Any] = {
+            "total_claims": len(self._claims),
+            "comparisons": [],
+            "evidence_scores": {},
+        }
+
+        if self._run is not None:
+            self._run = self._run.model_copy(
+                update={
+                    "metadata": {**self._run.metadata, "comparison": self._comparison_data},
+                    "updated_at": datetime.utcnow(),
+                }
+            )
+
+        return {
+            "success": True,
+            "data": self._comparison_data,
+        }
+
+    async def _step_synthesizing(self) -> dict[str, Any]:
+        """SYNTHESIZING: Erzeuge Synthese-Bericht mit SynthesisStage."""
+        self._transition_to("synthesizing")
+
+        try:
+            # LLM-Anfrage zählen
+            self._budget_tracker.increment_llm_requests(1)
+
+            if SynthesisStage is None:
+                logger.warning("SynthesisStage not available — generating fallback report")
+                return self._fallback_synthesis()
+
+            llm_provider = self._get_llm_provider()
+            if llm_provider is None:
+                logger.warning("No LLM provider for synthesis — generating fallback report")
+                return self._fallback_synthesis()
+
+            # Claims als Dicts für SynthesisStage
+            claims_dicts = [c.model_dump() for c in self._claims]
+
+            stage = SynthesisStage(
+                research_run_id=self._run.id if self._run else uuid4(),
+                llm_provider=llm_provider,
+                config=self._config,
+                claims=claims_dicts,
+            )
+
+            result = await stage.execute()
+
+            if result.success:
+                logger.info("Synthesis complete: report generated successfully")
+                report = result.data if result.data else {}
+                self._budget_tracker.record_time_elapsed()
+                return {
+                    "success": True,
+                    "data": report,
+                    "report": report,
+                }
+            else:
+                logger.warning("Synthesis returned success=False — generating fallback")
+                return self._fallback_synthesis_data(result.errors)
+
+        except Exception as exc:
+            logger.error("Synthesis failed: %s — generating fallback", exc)
+            return self._fallback_synthesis()
+
+    def _fallback_synthesis(self) -> dict[str, Any]:
+        """Fallback Synthese wenn SynthesisStage nicht verfügbar ist."""
+        logger.info("Generating fallback synthesis report")
+
+        fallback_report = {
+            "success": True,
+            "data": self._get_report(),
+            "report": self._get_report(),
+            "warning": "Fallback synthesis (LLM unavailable)",
+        }
+        self._budget_tracker.record_time_elapsed()
+        return fallback_report
+
+    def _fallback_synthesis_data(self, errors: list[str]) -> dict[str, Any]:
+        """Fallback report aus gescheiterter Synthese."""
+        report = self._get_report()
+        fallback = {
+            "success": True,
+            "data": report,
+            "report": report,
+            "synthesis_errors": errors,
+            "warning": "Synthesis degraded to fallback",
+        }
+        self._budget_tracker.record_time_elapsed()
+        return fallback
+
+    # ---------------------------------------------------------------
+    # Private: Helpers
+    # ---------------------------------------------------------------
+
+    def _create_research_run(self) -> ResearchRun:
+        """Erstelle eine ResearchRun-Instanz mit allen relevanten Feldern.
+
+        Returns
+        -------
+        ResearchRun
+            Das neue Run-Objekt.
+        """
+        budget_json = None
+        if self._budget_config is not None:
+            budget_json = self._budget_config.model_dump_json()
+
+        run = ResearchRun(
+            id=uuid4(),
+            research_id=self._research_id,
+            query=self._query,
+            state="created",
+            budget_config_json=budget_json,
+            metadata={
+                "depth": self._depth,
+                "created_by": "orchestrator",
+            },
+        )
+        self._run = run
+        return run
+
+    def _get_report(self) -> dict[str, Any]:
+        """Erzeuge den finalen Report aus allen Zwischenspeichern."""
+        report: dict[str, Any] = {
+            "run_id": self._run.id if self._run else None,
+            "query": self._query,
+            "state": self._state_machine.current_state.value,
+            "plan": self._plan,
+            "search_results": self._search_results,
+            "sources": self._sources,
+            "claims": [c.model_dump() for c in self._claims],
+            "claim_count": len(self._claims),
+            "source_count": len(self._sources),
+            "budget_usage": self._budget_tracker.get_usage(),
+            "duration_seconds": (
+                time.monotonic() - self._budget_tracker._start_time
+                if hasattr(self._budget_tracker, "_start_time")
+                else 0
+            ),
+        }
+
+        if self._comparison_data:
+            report["comparison"] = self._comparison_data
+        if self._run is not None:
+            report["metadata"] = self._run.metadata
+
+        return report
+
+    def _get_fallback_plan(self) -> dict[str, Any]:
+        """Minimaler Fallback-Plan falls der Planner komplett versagt."""
+        return {
+            "topic": self._query[:120],
+            "queries": [
+                {
+                    "query": self._query,
+                    "purpose": "Falls Planner ausgefallen: Basis-Query",
+                    "category": "general",
+                    "language": "de",
+                }
+            ],
+            "fallback": True,
+            "error": "Planner completely failed — using minimal fallback",
+        }
+
+    # ---------------------------------------------------------------
+    # Lifecycle
+    # ---------------------------------------------------------------
+
+    async def reset(self) -> None:
+        """Setze den Orchestrator zurück (zustand → CREATED)."""
+        self._state_machine.reset(ResearchRunState.CREATED)
+        self._plan = None
+        self._search_results = []
+        self._sources = []
+        self._claims = []
+        self._planner = None
+        self._crawler = None
+        self._multi_search = None
+        self._llm_provider = None
+        self._budget_tracker = BudgetTracker(self._budget_config)
+        logger.info("Orchestrator reset to CREATED state")

src/nsct/orchestration/state.py

@@ -0,0 +1,206 @@
+"""
+State Machine Model für den NSCT Research Orchestrator.
+
+Definiert die Lebenszyklus-Zustände eines Research-Run und validiert
+erlaubte Zustandübergänge entlang der Pipeline.
+"""
+
+from __future__ import annotations
+
+import logging
+from enum import Enum
+from typing import ClassVar, FrozenSet
+
+logger = logging.getLogger(__name__)
+
+
+class ResearchRunState(str, Enum):
+    """Mögliche Zustände eines Research-Laufs entlang der Pipeline."""
+
+    # --- Initialisierung ---
+    CREATED = "created"
+
+    # --- Pipeline-Schritte ---
+    PLANNING = "planning"
+    SEARCHING = "searching"
+    FETCHING = "fetching"
+    EXTRACTING = "extracting"
+    ANALYZING = "analyzing"
+    EXPANDING = "expanding"
+    COMPARING = "comparing"
+    SYNTHESIZING = "synthesizing"
+
+    # --- Endzustände ---
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+# Übergangsmatrix: Quelle -> zulässige Ziele (frozenset für Immutable)
+_VALID_TRANSITIONS: dict[ResearchRunState, FrozenSet[ResearchRunState]] = {
+    ResearchRunState.CREATED: frozenset(
+        (ResearchRunState.PLANNING,)
+    ),
+    ResearchRunState.PLANNING: frozenset(
+        (ResearchRunState.SEARCHING, ResearchRunState.FAILED)
+    ),
+    ResearchRunState.SEARCHING: frozenset(
+        (ResearchRunState.FETCHING, ResearchRunState.FAILED)
+    ),
+    ResearchRunState.FETCHING: frozenset(
+        (ResearchRunState.EXTRACTING, ResearchRunState.FAILED)
+    ),
+    ResearchRunState.EXTRACTING: frozenset(
+        (ResearchRunState.ANALYZING, ResearchRunState.FAILED)
+    ),
+    ResearchRunState.ANALYZING: frozenset(
+        (
+            ResearchRunState.EXPANDING,
+            ResearchRunState.COMPARING,
+            ResearchRunState.FAILED,
+        )
+    ),
+    ResearchRunState.EXPANDING: frozenset(
+        (
+            ResearchRunState.ANALYZING,  # zurück zur Analyse nach Expansion
+            ResearchRunState.COMPARING,
+            ResearchRunState.FAILED,
+        )
+    ),
+    ResearchRunState.COMPARING: frozenset(
+        (ResearchRunState.SYNTHESIZING, ResearchRunState.FAILED)
+    ),
+    ResearchRunState.SYNTHESIZING: frozenset(
+        (ResearchRunState.COMPLETED, ResearchRunState.FAILED)
+    ),
+    # Terminal-States: keine ausgehenden Übergänge.
+    ResearchRunState.COMPLETED: frozenset(),
+    ResearchRunState.FAILED: frozenset(),
+    ResearchRunState.CANCELLED: frozenset(),
+}
+
+
+class StateMachine:
+    """Endlicher Automat für den Lebenszyklus eines Research-Runs.
+
+    Parameters
+    ----------
+    initial_state : ResearchRunState
+        Der Startzustand der Maschine (default: CREATED).
+
+    Attributes
+    ----------
+    current_state : ResearchRunState
+        Der aktuell eingestellte Zustand.
+    """
+
+    # Die erlaubten Übergänge sind klassengebunden und typstarr.
+    _TRANSITIONS: ClassVar[dict[ResearchRunState, FrozenSet[ResearchRunState]]] = _VALID_TRANSITIONS
+
+    def __init__(self, initial_state: ResearchRunState = ResearchRunState.CREATED) -> None:
+        self.current_state: ResearchRunState = initial_state
+
+    def transition(self, new_state: str) -> bool:
+        """Führt einen Zustandübergang aus, wenn er erlaubt ist.
+
+        Parameters
+        ----------
+        new_state : str
+            Der gewünschte Zielzustand als String (z. B. ``"planning"``).
+
+        Returns
+        -------
+        bool
+            ``True`` wenn der Übergang erfolgreich war, ``False`` sonst
+            (z. B. bei einem unerlaubten Übergang oder Terminal-State).
+        """
+        target: ResearchRunState | None = None
+        try:
+            target = ResearchRunState(new_state)
+        except ValueError:
+            logger.info(
+                "Unknown state '%s' — transition %s -> %s rejected",
+                self.current_state.value,
+                new_state,
+            )
+            return False
+
+        return self._apply(self.current_state, target)
+
+    def is_terminal(self) -> bool:
+        """Prüft, ob die Maschine sich in einem Terminal-State befindet."""
+        return self.current_state in (
+            ResearchRunState.COMPLETED,
+            ResearchRunState.FAILED,
+            ResearchRunState.CANCELLED,
+        )
+
+    def reset(self, state: ResearchRunState | None = None) -> None:
+        """Setzt die Maschine zurück (z. B. vor einem neuen Run).
+
+        Parameters
+        ----------
+        state : ResearchRunState | None
+            Zielzustand; default: CREATED.
+        """
+        self.current_state = state or ResearchRunState.CREATED
+
+    def _apply(self, from_state: ResearchRunState, to_state: ResearchRunState) -> bool:
+        """Interne Übergangslogik mit Logging.
+
+        Parameters
+        ----------
+        from_state : ResearchRunState
+            Der aktuelle Zustand (zur Validierung).
+        to_state : ResearchRunState
+            Der gewünschte neue Zustand.
+
+        Returns
+        -------
+        bool
+            ``True`` wenn der Übergang ausgeführt wurde, ``False`` sonst.
+        """
+        allowed: FrozenSet[ResearchRunState] = _VALID_TRANSITIONS.get(from_state)
+        if allowed is None:
+            raise ValueError(
+                f"No outgoing transitions defined for state {from_state.value!r}"
+            )
+
+        # Terminal-States blockieren jeglichen Ausgang
+        if from_state in (
+            ResearchRunState.COMPLETED,
+            ResearchRunState.FAILED,
+            ResearchRunState.CANCELLED,
+        ):
+            logger.info(
+                "Terminal state %s reached — transition %s -> %s rejected",
+                from_state.value,
+                from_state.value,
+                to_state.value,
+            )
+            return False
+
+        # Selber-Stehen ist kein Fehler
+        if from_state == to_state:
+            logger.info(
+                "No-op: state already %s — skipping transition",
+                from_state.value,
+            )
+            return False
+
+        if to_state not in allowed:
+            logger.info(
+                "Transition %s -> %s rejected — not allowed",
+                from_state.value,
+                to_state.value,
+            )
+            return False
+
+        # Erlaubt: tatsächlich durchführen
+        self.current_state = to_state  # type: ignore[assignment]  # instance attr
+        logger.info(
+            "Transition %s -> %s accepted",
+            from_state.value,
+            to_state.value,
+        )
+        return True