diff --git a/src/nsct/orchestration/__init__.py b/src/nsct/orchestration/__init__.py index feb9665..2da4033 100644 --- a/src/nsct/orchestration/__init__.py +++ b/src/nsct/orchestration/__init__.py @@ -6,10 +6,19 @@ from .context_budget import ( ContextBudgetTracker, ) from .models import ResearchRun +from .state import ResearchRunState, StateMachine +from .orchestrator import ResearchOrchestrator +from ..provenance import ProvenanceEntry, ProvenanceLog, compute_research_run_hash __all__ = [ "ContextBudgetConfig", "ContextBudgetExhaustedError", "ContextBudgetTracker", "ResearchRun", + "ResearchRunState", + "StateMachine", + "ResearchOrchestrator", + "ProvenanceEntry", + "ProvenanceLog", + "compute_research_run_hash", ] \ No newline at end of file diff --git a/src/nsct/orchestration/models.py b/src/nsct/orchestration/models.py index 5a4f0ba..ba6eb67 100644 --- a/src/nsct/orchestration/models.py +++ b/src/nsct/orchestration/models.py @@ -65,5 +65,11 @@ class ResearchRun(BaseModel): default=0, description="Number of claims that were extracted.", ) + research_run_hash: str = Field( + default="", + min_length=64, + max_length=64, + description="Deterministic SHA256 hash of query + budget + created_at (Stage 21).", + ) model_config = {"frozen": True} \ No newline at end of file diff --git a/src/nsct/orchestration/orchestrator.py b/src/nsct/orchestration/orchestrator.py index fc07c21..74e3019 100644 --- a/src/nsct/orchestration/orchestrator.py +++ b/src/nsct/orchestration/orchestrator.py @@ -39,6 +39,7 @@ from nsct.orchestration.models import ResearchRun from nsct.orchestration.state import ResearchRunState, StateMachine from nsct.providers.abstract import MultiProviderSearch, SearchProvider from nsct.providers.priority_queue import Priority +from nsct.provenance import ProvenanceEntry, ProvenanceLog, compute_research_run_hash try: from nsct.stages.stage9_synthesis import SynthesisStage @@ -125,6 +126,10 @@ class ResearchOrchestrator: self._multi_search: MultiProviderSearch | None = None self._llm_provider = None + # Provenance (Stage 21) + self._provenance: ProvenanceLog | None = None + self._research_run_hash: str = "" + # --------------------------------------------------------------- # Properties # --------------------------------------------------------------- @@ -326,7 +331,7 @@ class ResearchOrchestrator: """Erstelle den ResearchRun und initialisiere die Pipeline. Setzt State auf CREATED, erstellt ResearchRun-Instanz, - initialisiert Budget-Tracker und State Machine. + initialisiert Budget-Tracker, State Machine und Provenance. Returns ------- @@ -341,11 +346,39 @@ class ResearchOrchestrator: self._run = self._create_research_run() self._transition_to("created") + # Provenance initialisieren (Stage 21) + budget_dict = self._budget_config.model_dump() + self._research_run_hash = compute_research_run_hash( + self._query, budget_dict, self._run.created_at.isoformat() + ) + self._provenance = ProvenanceLog(research_run_hash=self._research_run_hash) + + # Start-Eintrag + start_entry = ProvenanceEntry( + step="pipeline_start", + timestamp=datetime.now(timezone.utc), + inputs={ + "query": self._query, + "depth": self._depth, + "created_at": self._run.created_at.isoformat(), + }, + outputs={ + "run_id": str(self._run.id), + "research_run_hash": self._research_run_hash, + }, + metadata={ + "stage": "21", + "feature": "reproducibility", + }, + ) + self._provenance.add(start_entry) + logger.info( - "Research run started: id=%s, query=%s, depth=%s", + "Research run started: id=%s, query=%s, depth=%s, hash=%s", self._run.id, self._query[:60], self._depth, + self._research_run_hash, ) return self._run @@ -905,6 +938,126 @@ class ResearchOrchestrator: "error": "Planner completely failed — using minimal fallback", } + # --------------------------------------------------------------- + # Stage 21 — Provenance helpers + # --------------------------------------------------------------- + + def _log_step_entry(self, step_name: str) -> None: + """Logge Provenance-Eintrag bei Schritt-Start. + + Parameters + ---------- + step_name : str + Name des Schritts (z.B. "planning"). + """ + if self._provenance is None: + return + entry = ProvenanceEntry( + step=f"stage_{step_name}", + timestamp=datetime.now(timezone.utc), + inputs={ + "step": step_name, + "run_id": str(self._run.id) if self._run else None, + }, + outputs={}, + ) + self._provenance.add(entry) + + def _log_step_exit(self, step_name: str, result: dict[str, Any]) -> None: + """Logge Provenance-Eintrag bei Schritt-Ende. + + Parameters + ---------- + step_name : str + Name des Schritts. + result : dict + Ergebnis-Dict vom Schritt. + """ + if self._provenance is None: + return + success = result.get("success", False) + outputs: dict[str, Any] = { + "success": success, + "step": step_name, + } + # Extrahiere nützliche Counts/Referenzen + for key in ("plan", "url_count", "source_count", "claim_count"): + if key in result: + outputs[key] = result[key] + if "data" in result and isinstance(result["data"], dict): + data = result["data"] + if "plan" in data and isinstance(data["plan"], dict): + outputs["plan_topics"] = data["plan"].get("topic", "")[:80] + if "urls" in data: + outputs["url_count"] = len(data["urls"]) + if "sources" in data: + outputs["source_count"] = len(data["sources"]) + if "claims" in data: + outputs["claim_count"] = len(data["claims"]) + + error_str = result.get("error", None) + if error_str is None and not success: + error_str = f"step_failed" + + entry = ProvenanceEntry( + step=f"stage_{step_name}", + timestamp=datetime.now(timezone.utc), + inputs={"step": step_name}, + outputs=outputs, + error=error_str, + ) + self._provenance.add(entry) + + def get_provenance(self) -> dict[str, Any]: + """Gibt den aktuellen Provenance-Log als Dict zurück. + + Returns + ------- + dict + Vollständiger Provenance-Log mit research_run_hash, entries und total_steps. + """ + if self._provenance is None: + return ProvenanceLog(research_run_hash=self._research_run_hash).to_dict() + return self._provenance.to_dict() + + async def save_provenance_to_db(self, db_session, research_run_id: str) -> int: + """Serialisiere alle Provenance-Einträge und speichere sie in der DB. + + Parameters + ---------- + db_session : sqlalchemy.AsyncSession + Datenbank-Sitzung. + research_run_id : str + UUID als String fuer die Zuordnung. + + Returns + ------- + int + Anzahl gespeicherter Einträge. + """ + if self._provenance is None: + return 0 + + from nsct.storage.models import ResearchRunProvenanceModel + + saved = 0 + for entry in self._provenance.entries: + row = ResearchRunProvenanceModel( + research_run_id=research_run_id, + research_run_hash=self._research_run_hash, + step=entry.step, + timestamp=entry.timestamp, + inputs_json=entry.inputs, + outputs_json=entry.outputs, + error_json=entry.error, + metadata_json=entry.metadata, + ) + db_session.add(row) + saved += 1 + + await db_session.commit() + return saved + # --------------------------------------------------------------- # Lifecycle # --------------------------------------------------------------- diff --git a/src/nsct/provenance.py b/src/nsct/provenance.py new file mode 100644 index 0000000..a94b95f --- /dev/null +++ b/src/nsct/provenance.py @@ -0,0 +1,162 @@ +"""Provenance-Modul für NSCT — research_run_hash und Provenance-Tracking (Stage 21). + +Berechnet deterministische SHA256-Hashes fuer Research-Runs und trackt +schrittweise Provenance-Einträge mit timestamps, inputs und outputs. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, field +from datetime import datetime, timezone + + +def compute_research_run_hash(query: str, budget_config: dict, created_at: str) -> str: + """Berechne einen deterministischen SHA256-Hash fuer einen Research-Run. + + Der Hash wird aus drei deterministischen Komponenten zusammengesetzt: + 1. Query (stripped, trim) + 2. Budget-Config (sortierte JSON-Serialisierung) + 3. created_at (ISO-String) + + Parameters + ---------- + query : str + Die Forschungsfrage / Query. + budget_config : dict + Budget-Konfiguration als Dict. + created_at : str + ISO-String des Erstellungszeitpunkts. + + Returns + ------- + str + 64 Zeichen langer Hex-Hash (SHA256). + """ + stripped_query = query.strip() + + budget_json = json.dumps(budget_config, sort_keys=True, ensure_ascii=False) + + raw = f"{stripped_query}|{budget_json}|{created_at}" + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class ProvenanceEntry: + """Ein einzelner Provenance-Eintrag fuer einen Pipeline-Schritt. + + Felder + ------ + step : str + Name des Schritts (z.B. "stage4_planning"). + timestamp : datetime + Zeitpunkt des Eintrags (timezone.utc). + inputs : dict + Minimale Eingabe-Daten. + outputs : dict + Ausgabe-Daten (Referenzen, Counts). + metadata : dict + Beliebige Zusatzinfos. + error : str | None + Fehlermeldung, falls der Schritt fehlschlug. + """ + + step: str + timestamp: datetime + inputs: dict + outputs: dict + metadata: dict = field(default_factory=dict) + error: str | None = None + + def to_dict(self) -> dict: + """Konvertiere den Eintrag in ein serialisierbares Dict. + + Returns + ------- + dict + Serialisierbarer Dict repraesentation. + """ + return { + "step": self.step, + "timestamp": self.timestamp.isoformat() if self.timestamp else None, + "inputs": self.inputs, + "outputs": self.outputs, + "metadata": self.metadata, + "error": self.error, + } + + +@dataclass +class ProvenanceLog: + """Sammelt Provenance-Einträge fuer einen Research-Run. + + Felder + ------ + research_run_hash : str + Der deterministische Hash des Runs. + entries : list[ProvenanceEntry] + Liste aller Provenance-Einträge. + """ + + research_run_hash: str + entries: list[ProvenanceEntry] = field(default_factory=list) + + def add(self, entry: ProvenanceEntry) -> None: + """Fuege einen Provenance-Eintrag hinzu. + + Parameters + ---------- + entry : ProvenanceEntry + Der hinzuzufuegende Eintrag. + """ + self.entries.append(entry) + + def to_dict(self) -> dict: + """Konvertiere den gesamten Log in ein serialisierbares Dict. + + Returns + ------- + dict + Full provenance log mit allen Eintraegen und Metadaten. + """ + return { + "research_run_hash": self.research_run_hash, + "entries": [e.to_dict() for e in self.entries], + "total_steps": len(self.entries), + } + + def get_entries_by_step(self, step: str) -> list[ProvenanceEntry]: + """Filtere Eintraege nach Schritt-Name. + + Parameters + ---------- + step : str + Name des Schritts zum Filtern. + + Returns + ------- + list[ProvenanceEntry] + Alle Eintraege fuer den gegebenen Schritt. + """ + return [e for e in self.entries if e.step == step] + + def get_last_entry(self) -> ProvenanceEntry | None: + """Gib den letzten Eintrag zurueck. + + Returns + ------- + ProvenanceEntry | None + Der letzte Eintrag oder None wenn leer. + """ + return self.entries[-1] if self.entries else None + + def has_entries(self) -> bool: + """Pruefe ob Eintraege vorhanden sind. + + Returns + ------- + bool + True wenn mindestens ein Eintrag existiert. + """ + return len(self.entries) > 0 \ No newline at end of file diff --git a/src/nsct/storage/models.py b/src/nsct/storage/models.py index 6192445..7afd276 100644 --- a/src/nsct/storage/models.py +++ b/src/nsct/storage/models.py @@ -754,4 +754,30 @@ class AudioClaimModel(Base): Index("ix_audio_claims_transcript_id", "transcript_id"), Index("ix_audio_claims_claim_type", "claim_type"), Index("ix_audio_claims_speaker_id", "speaker_id"), + ) + + +# --------------------------------------------------------------------------- +# Stage 21 — Research Run Provenance (Reproduzierbarkeit) +# --------------------------------------------------------------------------- + + +class ResearchRunProvenanceModel(Base): + """Provenance-Table für vollständige Nachverfolgbarkeit aller Schritte (Stage 21).""" + + __tablename__ = "research_run_provenance" + + id = Column(BigInteger, primary_key=True, autoincrement=True) + research_run_id = Column(String(36), nullable=False, index=True) + research_run_hash = Column(String(64), nullable=False, index=True) + step = Column(String(128), nullable=False) + timestamp = Column(DateTime, nullable=False, default=datetime.utcnow) + inputs_json = Column(JSON, nullable=True) + outputs_json = Column(JSON, nullable=True) + error_json = Column(JSON, nullable=True) + metadata_json = Column(JSON, nullable=True) + + __table_args__ = ( + Index("ix_provenance_run_step", "research_run_id", "step"), + Index("ix_provenance_run_timestamp", "research_run_id", "timestamp"), ) \ No newline at end of file diff --git a/tests/stages/test_provenance.py b/tests/stages/test_provenance.py new file mode 100644 index 0000000..95f5624 --- /dev/null +++ b/tests/stages/test_provenance.py @@ -0,0 +1,442 @@ +"""Tests für Stage 21 — Provenance & research_run_hash (Reproduzierbarkeit).""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from uuid import UUID, uuid4 + +import pytest + +from nsct.provenance import ProvenanceEntry, ProvenanceLog, compute_research_run_hash +from nsct.orchestration.models import ResearchRun + + +# =================================================================== +# compute_research_run_hash — Determinismus & Korrektheit +# =================================================================== + + +class TestComputeResearchRunHash: + """Tests für compute_research_run_hash().""" + + def test_deterministic_same_input(self) -> None: + """Gleiche Inputs → gleicher Hash.""" + query = "Wie wirkt sich KI auf den Arbeitsmarkt?" + budget = {"max_llm_requests": 100, "max_sources": 50} + created_at = "2025-01-15T10:30:00+00:00" + + h1 = compute_research_run_hash(query, budget, created_at) + h2 = compute_research_run_hash(query, budget, created_at) + + assert h1 == h2 + assert len(h1) == 64 + + def test_different_query_different_hash(self) -> None: + """Andere Query → anderer Hash.""" + budget = {"max_llm_requests": 100} + created_at = "2025-01-15T10:30:00+00:00" + + h1 = compute_research_run_hash("Query A", budget, created_at) + h2 = compute_research_run_hash("Query B", budget, created_at) + + assert h1 != h2 + + def test_different_budget_different_hash(self) -> None: + """Andere Budget → anderer Hash.""" + query = "Test query" + created_at = "2025-01-15T10:30:00+00:00" + + budget1 = {"max_llm_requests": 50} + budget2 = {"max_llm_requests": 200} + + h1 = compute_research_run_hash(query, budget1, created_at) + h2 = compute_research_run_hash(query, budget2, created_at) + + assert h1 != h2 + + def test_different_created_at_different_hash(self) -> None: + """Andere created_at → anderer Hash.""" + query = "Test query" + budget = {"max_llm_requests": 100} + + h1 = compute_research_run_hash(query, budget, "2025-01-15T10:30:00+00:00") + h2 = compute_research_run_hash(query, budget, "2025-01-15T11:30:00+00:00") + + assert h1 != h2 + + def test_hash_is_sha256_hex_64_chars(self) -> None: + """Hash ist 64 Zeichen Hex.""" + h = compute_research_run_hash( + "Test", + {"max_llm_requests": 1}, + "2025-01-01T00:00:00+00:00", + ) + assert len(h) == 64 + int(h, 16) # Sollte keine Exception werfen + + def test_hash_sha256_verification(self) -> None: + """Hash stimmt mit manuell berechnetem SHA256 überein.""" + query = "Wie wirkt sich KI auf den Arbeitsmarkt?" + budget = {"max_llm_requests": 100, "max_sources": 50} + created_at = "2025-01-15T10:30:00+00:00" + + expected_raw = f"{query}|{json.dumps(budget, sort_keys=True, ensure_ascii=False)}|{created_at}" + expected = hashlib.sha256(expected_raw.encode("utf-8")).hexdigest() + + assert compute_research_run_hash(query, budget, created_at) == expected + + def test_query_stripped(self) -> None: + """Whitespace am Ende der Query wird gestripped.""" + budget = {"max_llm_requests": 100} + created_at = "2025-01-01T00:00:00+00:00" + + h1 = compute_research_run_hash(" query ", budget, created_at) + h2 = compute_research_run_hash("query", budget, created_at) + + assert h1 == h2 + + def test_empty_inputs(self) -> None: + """Leere Inputs ergeben einen deterministischen Hash.""" + h1 = compute_research_run_hash("", {}, "") + h2 = compute_research_run_hash("", {}, "") + + assert h1 == h2 + assert len(h1) == 64 + + def test_unicode_in_query(self) -> None: + """Unicode-Characters in Query werden korrekt behandelt.""" + budget = {} + created_at = "2025-01-01T00:00:00+00:00" + + h1 = compute_research_run_hash("Überprüfung ñ测试", budget, created_at) + h2 = compute_research_run_hash("Überprüfung ñ测试", budget, created_at) + + assert h1 == h2 + assert len(h1) == 64 + + def test_budget_order_independence(self) -> None: + """Budget-Dict Reihenfolge beeinflusst Hash nicht (sort_keys=True).""" + query = "Test" + created_at = "2025-01-01T00:00:00+00:00" + + budget1 = {"a": 1, "b": 2, "c": 3} + budget2 = {"c": 3, "a": 1, "b": 2} + + h1 = compute_research_run_hash(query, budget1, created_at) + h2 = compute_research_run_hash(query, budget2, created_at) + + assert h1 == h2 + + +# =================================================================== +# ProvenanceEntry +# =================================================================== + + +class TestProvenanceEntry: + """Tests für ProvenanceEntry.""" + + def test_basic_entry(self) -> None: + """Basis-Eintrag mit allen Feldern.""" + ts = datetime(2025, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + entry = ProvenanceEntry( + step="stage4_planning", + timestamp=ts, + inputs={"query": "Test"}, + outputs={"plan": "minimal"}, + ) + + assert entry.step == "stage4_planning" + assert entry.timestamp == ts + assert entry.inputs == {"query": "Test"} + assert entry.outputs == {"plan": "minimal"} + assert entry.metadata == {} + assert entry.error is None + + def test_to_dict(self) -> None: + """to_dict() erzeugt korrektes Dict.""" + ts = datetime(2025, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + entry = ProvenanceEntry( + step="stage5_extracting", + timestamp=ts, + inputs={"url_count": 5}, + outputs={"claim_count": 12}, + metadata={"provider": "qwen3"}, + error=None, + ) + + d = entry.to_dict() + + assert d["step"] == "stage5_extracting" + assert d["timestamp"] == "2025-01-15T10:30:00+00:00" + assert d["inputs"] == {"url_count": 5} + assert d["outputs"] == {"claim_count": 12} + assert d["metadata"] == {"provider": "qwen3"} + assert d["error"] is None + + def test_to_dict_with_error(self) -> None: + """to_dict() mit Fehlermeldung.""" + entry = ProvenanceEntry( + step="stage6_fetching", + timestamp=datetime.now(timezone.utc), + inputs={"url": "https://example.com"}, + outputs={}, + error="Connection timeout", + ) + + d = entry.to_dict() + assert d["error"] == "Connection timeout" + + def test_default_metadata(self) -> None: + """metadata ist standardmäßig leer.""" + entry = ProvenanceEntry( + step="test", + timestamp=datetime.now(timezone.utc), + inputs={}, + outputs={}, + ) + assert entry.metadata == {} + + +# =================================================================== +# ProvenanceLog +# =================================================================== + + +class TestProvenanceLog: + """Tests für ProvenanceLog.""" + + def test_empty_log(self) -> None: + """Leerer Log mit 0 Einträgen.""" + log = ProvenanceLog(research_run_hash="a" * 64) + assert log.entries == [] + assert not log.has_entries() + + def test_add_entry(self) -> None: + """Eintrag hinzufügen und via add() prüfen.""" + log = ProvenanceLog(research_run_hash="b" * 64) + entry = ProvenanceEntry( + step="stage4_planning", + timestamp=datetime.now(timezone.utc), + inputs={"query": "test"}, + outputs={"plan": "x"}, + ) + log.add(entry) + + assert len(log.entries) == 1 + assert log.has_entries() + assert log.entries[0].step == "stage4_planning" + + def test_to_dict_empty(self) -> None: + """to_dict() auf leerem Log.""" + log = ProvenanceLog(research_run_hash="c" * 64) + d = log.to_dict() + + assert d["research_run_hash"] == "c" * 64 + assert d["entries"] == [] + assert d["total_steps"] == 0 + + def test_to_dict_with_entries(self) -> None: + """to_dict() mit Einträgen.""" + ts = datetime(2025, 1, 15, 10, 0, 0, tzinfo=timezone.utc) + log = ProvenanceLog(research_run_hash="d" * 64) + log.add(ProvenanceEntry(step="planning", timestamp=ts, inputs={}, outputs={"plan": "x"})) + log.add(ProvenanceEntry(step="searching", timestamp=ts, inputs={}, outputs={"urls": 5})) + + d = log.to_dict() + + assert d["total_steps"] == 2 + assert len(d["entries"]) == 2 + assert d["entries"][0]["step"] == "planning" + assert d["entries"][1]["step"] == "searching" + + def test_get_entries_by_step(self) -> None: + """Filtere Einträge nach Schritt.""" + ts = datetime.now(timezone.utc) + log = ProvenanceLog(research_run_hash="e" * 64) + log.add(ProvenanceEntry(step="stage_planning", timestamp=ts, inputs={}, outputs={})) + log.add(ProvenanceEntry(step="stage_searching", timestamp=ts, inputs={}, outputs={})) + log.add(ProvenanceEntry(step="stage_planning", timestamp=ts, inputs={}, outputs={})) + + results = log.get_entries_by_step("stage_planning") + assert len(results) == 2 + + def test_get_entries_by_step_no_match(self) -> None: + """Filter gibt leere Liste zurück wenn kein Match.""" + log = ProvenanceLog(research_run_hash="f" * 64) + assert log.get_entries_by_step("nonexistent") == [] + + def test_get_last_entry(self) -> None: + """Letzter Eintrag wird korrekt zurückgegeben.""" + ts = datetime.now(timezone.utc) + log = ProvenanceLog(research_run_hash="g" * 64) + log.add(ProvenanceEntry(step="first", timestamp=ts, inputs={}, outputs={})) + log.add(ProvenanceEntry(step="second", timestamp=ts, inputs={}, outputs={})) + + last = log.get_last_entry() + assert last is not None + assert last.step == "second" + + def test_get_last_entry_empty(self) -> None: + """Leerer Log → None.""" + log = ProvenanceLog(research_run_hash="h" * 64) + assert log.get_last_entry() is None + + def test_multiple_adds(self) -> None: + """Mehrfaches Hinzufügen zählt korrekt.""" + log = ProvenanceLog(research_run_hash="i" * 64) + for i in range(10): + log.add(ProvenanceEntry( + step=f"step_{i}", + timestamp=datetime.now(timezone.utc), + inputs={}, + outputs={}, + )) + assert len(log.entries) == 10 + assert log.to_dict()["total_steps"] == 10 + + +# =================================================================== +# ResearchRun Model — research_run_hash field +# =================================================================== + + +class TestResearchRunModel: + """Tests für ResearchRun mit research_run_hash.""" + + def test_research_run_hash_default_empty(self) -> None: + """research_run_hash ist standardmäßig leerer String.""" + run = ResearchRun( + query="Test query", + research_id=uuid4(), + ) + assert run.research_run_hash == "" + + def test_research_run_hash_set(self) -> None: + """research_run_hash kann gesetzt werden.""" + run = ResearchRun( + query="Test query", + research_id=uuid4(), + research_run_hash="a" * 64, + ) + assert run.research_run_hash == "a" * 64 + assert len(run.research_run_hash) == 64 + + def test_research_run_hash_length_validation(self) -> None: + """Zu kurzer/ langer Hash wirft ValidationError.""" + with pytest.raises(Exception): # pydantic.ValidationError + ResearchRun( + query="Test", + research_id=uuid4(), + research_run_hash="too_short", + ) + + +# =================================================================== +# Provenance-Tabellen-Model Validierung +# =================================================================== + + +class TestResearchRunProvenanceModel: + """Tests für ResearchRunProvenanceModel Schema.""" + + def test_model_tablename(self) -> None: + """Tabellenname ist korrekt.""" + from nsct.storage.models import ResearchRunProvenanceModel + + assert ResearchRunProvenanceModel.__tablename__ == "research_run_provenance" + + def test_model_columns_exist(self) -> None: + """Alle erwarteten Spalten existieren.""" + from nsct.storage.models import ResearchRunProvenanceModel + + column_names = {c.key for c in ResearchRunProvenanceModel.__table__.columns} + + expected = { + "id", + "research_run_id", + "research_run_hash", + "step", + "timestamp", + "inputs_json", + "outputs_json", + "error_json", + "metadata_json", + } + assert expected.issubset(column_names) + + def test_model_indexes(self) -> None: + """Indizes sind korrekt konfiguriert.""" + from nsct.storage.models import ResearchRunProvenanceModel + + index_names = {idx.name for idx in ResearchRunProvenanceModel.__table__.indexes} + + assert "ix_provenance_run_step" in index_names + assert "ix_provenance_run_timestamp" in index_names + + def test_model_primary_key(self) -> None: + """Primärschlüssel ist BigInteger autoincrement.""" + from nsct.storage.models import ResearchRunProvenanceModel + + pk = ResearchRunProvenanceModel.__table__.primary_key + pk_column = list(pk)[0] + assert pk_column.name == "id" + assert pk_column.autoincrement is True + + +# =================================================================== +# Orchestrator Provenance-Integration +# =================================================================== + + +class TestOrchestratorProvenanceIntegration: + """Tests für Provenance-Integration im Orchestrator.""" + + def test_orchestrator_has_provenance_import(self) -> None: + """Provenance-Module werden im Orchestrator importiert.""" + from nsct.orchestration.orchestrator import ResearchOrchestrator + + # Prüfe dass die Funktion importiert wurde + from nsct.provenance import compute_research_run_hash + assert compute_research_run_hash is not None + + def test_orchestrator_has_provenance_methods(self) -> None: + """Orchestrator hat Provenance-Methoden.""" + from nsct.orchestration.orchestrator import ResearchOrchestrator + + assert hasattr(ResearchOrchestrator, "get_provenance") + assert hasattr(ResearchOrchestrator, "save_provenance_to_db") + assert hasattr(ResearchOrchestrator, "_log_step_entry") + assert hasattr(ResearchOrchestrator, "_log_step_exit") + + def test_get_provenance_returns_dict(self) -> None: + """get_provenance() gibt Dict zurück.""" + from nsct.orchestration.orchestrator import ResearchOrchestrator + + assert callable(ResearchOrchestrator.get_provenance) + + def test_compute_hash_in_orchestrator_start(self, clean_env: None) -> None: + """Der Hash wird im start() berechnet und genutzt.""" + from nsct.orchestration.orchestrator import ResearchOrchestrator + from nsct.config import AppSettings + + settings = AppSettings.from_env() + orch = ResearchOrchestrator( + config=settings, + research_id=uuid4(), + query="Test query", + ) + + assert orch._research_run_hash == "" + assert orch._provenance is None + + def test_research_run_hash_in_models_export(self) -> None: + """ResearchRun mit hash kann importiert werden.""" + run = ResearchRun( + query="Test", + research_id=uuid4(), + research_run_hash="x" * 64, + ) + assert run.research_run_hash == "x" * 64 \ No newline at end of file