Spaces:
Running
Running
| """Read-only research registry and report source for the dashboard. | |
| Exposes the MHRN Scientific Evidence Framework (B5D-SEF) artifacts | |
| located under ``research/`` to the operator dashboard. The source is | |
| strictly read-only and resolves paths defensively so a missing or | |
| partially populated research tree never crashes the dashboard. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import re | |
| import time | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| from typing import Any, cast | |
| from .models import JSONValue | |
| class ResearchDocument: | |
| """One research artifact exposed to the dashboard.""" | |
| name: str | |
| path: str | |
| kind: str | |
| size_bytes: int | |
| category: str | |
| def classify_ai_operation(manifest: dict[str, Any]) -> str: | |
| """Classify AI execution provenance for read-only dashboard display.""" | |
| explicit = manifest.get("ai_operation_mode", manifest.get("operation_mode")) | |
| if isinstance(explicit, str) and explicit in { | |
| "REPLAY", | |
| "LIVE_FROZEN_MODEL", | |
| "LIVE_EXTERNAL_API", | |
| "NONE", | |
| "UNKNOWN", | |
| }: | |
| return explicit | |
| interactions_value: object = manifest.get("ai_interactions", []) | |
| interactions: list[Any] = [] | |
| if isinstance(interactions_value, list): | |
| interactions.extend( | |
| interactions_value # pyright: ignore[reportUnknownArgumentType] | |
| ) | |
| if not interactions and isinstance(manifest.get("ai_model_provenance"), dict): | |
| interactions = [{"model_provenance": manifest["ai_model_provenance"]}] | |
| if not interactions: | |
| return "NONE" | |
| network_mode = str(manifest.get("network_mode", "")).upper() | |
| for interaction in interactions: | |
| if not isinstance(interaction, dict): | |
| continue | |
| interaction_data = cast(dict[str, Any], interaction) | |
| provenance_value: Any = interaction_data.get("model_provenance", {}) | |
| provenance = ( | |
| cast(dict[str, Any], provenance_value) | |
| if isinstance(provenance_value, dict) | |
| else {} | |
| ) | |
| if provenance: | |
| provider = str( | |
| provenance.get("provider", provenance.get("backend", "")) | |
| ).lower() | |
| if "replay" in provider or provenance.get("replay") is True: | |
| return "REPLAY" | |
| if network_mode in {"OFFLINE", "FROZEN_CORPUS"} and ( | |
| provider in {"ollama", "local", "frozen_model"} or "frozen" in provider | |
| ): | |
| return "LIVE_FROZEN_MODEL" | |
| if network_mode == "LIVE_NETWORK" or ( | |
| provider and provider not in {"unknown", "not_reported"} | |
| ): | |
| return "LIVE_EXTERNAL_API" | |
| return "UNKNOWN" | |
| def classify_research_operation_status(manifest: dict[str, Any]) -> str: | |
| """Classify the scientific AI operating status for read-only display.""" | |
| exposure = str(manifest.get("ai_exposure", "")).lower() | |
| taint = str(manifest.get("causal_taint", "")).upper() | |
| if exposure == "none" and taint == "PURE": | |
| return "PURE EXPERIMENT" | |
| if taint == "AI_INFLUENCED" or exposure in { | |
| "bounded_controller", | |
| "adaptive_controller", | |
| }: | |
| return "AI CAUSALLY ACTIVE" | |
| if taint == "PROPOSED" or exposure == "advisor": | |
| return "AI PROPOSING" | |
| if taint == "OBSERVED" or exposure in {"observer_only", "semantic_interface"}: | |
| return "AI OBSERVING" | |
| return "UNKNOWN" | |
| def _experiment_date_from_id(experiment_id: str) -> str | None: | |
| """Infer an ISO date only when the experiment ID carries YYYYMMDD explicitly.""" | |
| match = re.search(r"(20\d{6})(?:$|[-_])", experiment_id) | |
| if match is None: | |
| return None | |
| raw = match.group(1) | |
| return f"{raw[:4]}-{raw[4:6]}-{raw[6:8]}" | |
| def normalize_experiment_manifest( | |
| directory: Path, manifest: dict[str, Any] | |
| ) -> dict[str, Any]: | |
| """Project heterogeneous immutable manifests into one dashboard read model. | |
| The returned mapping is a copy. Canonical experiment files are never rewritten. | |
| This keeps older workflow manifests and newer source-bound experiment packages | |
| equally reviewable in the dashboard. | |
| """ | |
| normalized = dict(manifest) | |
| experiment_id = str(normalized.get("experiment_id") or directory.name) | |
| normalized.setdefault("experiment_id", experiment_id) | |
| if not isinstance(normalized.get("research_questions"), list): | |
| question = normalized.get("research_question") | |
| if isinstance(question, str) and question: | |
| normalized["research_questions"] = [question] | |
| if not isinstance(normalized.get("hypotheses"), list): | |
| hypothesis = normalized.get("hypothesis") | |
| if isinstance(hypothesis, str) and hypothesis: | |
| normalized["hypotheses"] = [hypothesis] | |
| if not normalized.get("created_at"): | |
| timestamp = normalized.get("timestamp") or normalized.get("timestamp_utc") | |
| if isinstance(timestamp, str) and timestamp: | |
| normalized["created_at"] = timestamp | |
| else: | |
| inferred = _experiment_date_from_id(experiment_id) | |
| if inferred is not None: | |
| normalized["created_at"] = inferred | |
| if not normalized.get("experiment_status"): | |
| status = normalized.get("status") | |
| if isinstance(status, str) and status: | |
| normalized["experiment_status"] = status | |
| elif isinstance(normalized.get("results"), dict): | |
| result_status = cast(dict[str, Any], normalized["results"]).get("status") | |
| if isinstance(result_status, str) and result_status: | |
| normalized["experiment_status"] = ( | |
| "completed" | |
| if result_status | |
| in { | |
| "SUPPORTED_WITHIN_PREREGISTERED_PROTOCOL", | |
| "NO_PREDEFINED_TOPOLOGY_DIFFERENCE_DETECTED", | |
| } | |
| else result_status.lower() | |
| ) | |
| existing_artifacts = normalized.get("artifacts") | |
| artifacts: dict[str, Any] = ( | |
| dict(cast(dict[str, Any], existing_artifacts)) | |
| if isinstance(existing_artifacts, dict) | |
| else {} | |
| ) | |
| inferred_artifacts = { | |
| "report": "report.md", | |
| "summary": "summary.md", | |
| "statistics": "analysis/statistics.json", | |
| "raw_data": "data/evaluation.json", | |
| "calibration": "data/calibration.json", | |
| "review": "review_request.json", | |
| "manifest": "manifest.json", | |
| "checksums": "checksums.sha256", | |
| } | |
| for key, relative in inferred_artifacts.items(): | |
| if key not in artifacts and (directory / relative).is_file(): | |
| artifacts[key] = relative | |
| if artifacts: | |
| normalized["artifacts"] = artifacts | |
| results = normalized.get("results") | |
| if isinstance(results, dict): | |
| result_status = cast(dict[str, Any], results).get("status") | |
| if isinstance(result_status, str) and result_status: | |
| normalized.setdefault("scientific_status", result_status) | |
| return normalized | |
| class ResearchSource: | |
| """Read-only view over the ``research/`` directory tree. | |
| Categories: | |
| registry — YAML registry files (questions, hypotheses, ...) | |
| generated — auto-generated markdown reports | |
| experiments — experiment manifests | |
| literature — BibTeX databases | |
| schemas — JSON schemas | |
| """ | |
| _CATEGORIES: tuple[str, ...] = ( | |
| "registry", | |
| "generated", | |
| "experiments", | |
| "reports", | |
| "analysis", | |
| "benchmarks", | |
| "literature", | |
| "publications", | |
| "schemas", | |
| "external_review", | |
| "ethics", | |
| ) | |
| def __init__(self, research_root: Path) -> None: | |
| self._root = research_root.resolve() | |
| self._documents_cache: list[ResearchDocument] | None = None | |
| self._documents_cache_time: float = 0.0 | |
| self._cache_ttl: float = 5.0 # Sekunden | |
| def is_available(self) -> bool: | |
| """Return True when the research root exists and is a directory.""" | |
| return self._root.is_dir() | |
| def root(self) -> Path: | |
| return self._root | |
| def _invalidate_cache(self) -> None: | |
| """Force cache refresh on next call.""" | |
| self._documents_cache = None | |
| self._documents_cache_time = 0.0 | |
| def list_documents(self, max_count: int = 0) -> list[ResearchDocument]: | |
| """List research artifacts grouped by category, with caching. | |
| Args: | |
| max_count: Maximum number of documents to return. | |
| 0 = return all (uncached behavior). | |
| Returns: | |
| Sorted list of ResearchDocument objects. | |
| """ | |
| now = time.monotonic() | |
| # Use cache only when requesting all documents | |
| if max_count == 0 and self._documents_cache is not None: | |
| if now - self._documents_cache_time < self._cache_ttl: | |
| return self._documents_cache | |
| if not self.is_available(): | |
| return [] | |
| documents: list[ResearchDocument] = [] | |
| for category in self._CATEGORIES: | |
| directory = self._root / category | |
| if not directory.is_dir(): | |
| continue | |
| for entry in sorted(directory.rglob("*")): | |
| if not entry.is_file(): | |
| continue | |
| rel = entry.relative_to(self._root) | |
| documents.append( | |
| ResearchDocument( | |
| name=entry.name, | |
| path=str(rel).replace("\\", "/"), | |
| kind=entry.suffix.lstrip(".") or "file", | |
| size_bytes=entry.stat().st_size, | |
| category=category, | |
| ) | |
| ) | |
| # Early exit when max_count is set and reached | |
| if max_count > 0 and len(documents) >= max_count: | |
| break | |
| if max_count > 0 and len(documents) >= max_count: | |
| break | |
| if max_count == 0: | |
| self._documents_cache = documents | |
| self._documents_cache_time = now | |
| return documents | |
| def read_content(self, relative_path: str) -> str: | |
| """Read a research file as UTF-8 text with path-traversal protection.""" | |
| resolved = self._safe_resolve(relative_path) | |
| if resolved is None: | |
| raise FileNotFoundError(f"Research document not found: {relative_path}") | |
| return resolved.read_text(encoding="utf-8") | |
| def read_bytes(self, relative_path: str) -> bytes: | |
| resolved = self._safe_resolve(relative_path) | |
| if resolved is None: | |
| raise FileNotFoundError(f"Research document not found: {relative_path}") | |
| return resolved.read_bytes() | |
| def registry_summary(self) -> dict[str, JSONValue]: | |
| """Return a compact summary of registry counts for the dashboard.""" | |
| summary: dict[str, JSONValue] = { | |
| "available": self.is_available(), | |
| "root": str(self._root), | |
| "categories": {}, | |
| } | |
| if not self.is_available(): | |
| return summary | |
| counts: dict[str, JSONValue] = {} | |
| for category in self._CATEGORIES: | |
| directory = self._root / category | |
| if directory.is_dir(): | |
| counts[category] = sum( | |
| 1 for entry in directory.rglob("*") if entry.is_file() | |
| ) | |
| else: | |
| counts[category] = 0 | |
| summary["categories"] = counts | |
| return summary | |
| def generated_reports(self) -> list[dict[str, JSONValue]]: | |
| """List auto-generated markdown reports with metadata.""" | |
| directory = self._root / "generated" | |
| if not directory.is_dir(): | |
| return [] | |
| reports: list[dict[str, JSONValue]] = [] | |
| for entry in sorted(directory.glob("*.md")): | |
| reports.append( | |
| { | |
| "name": entry.stem, | |
| "path": str(entry.relative_to(self._root)).replace("\\", "/"), | |
| "size_bytes": entry.stat().st_size, | |
| } | |
| ) | |
| return reports | |
| def ai_reports( | |
| self, experiment_id: str | None = None | |
| ) -> list[dict[str, JSONValue]]: | |
| """List canonical AIRR JSON/Markdown pairs without exposing write access.""" | |
| reports: list[dict[str, JSONValue]] = [] | |
| directories = [ | |
| directory | |
| for directory in (self._root / "experiments").glob("*/reports") | |
| if directory.is_dir() | |
| ] | |
| legacy_directory = self._root / "reports" | |
| if legacy_directory.is_dir(): | |
| directories.append(legacy_directory) | |
| for directory in directories: | |
| current_experiment = directory.parent.name | |
| for entry in sorted(directory.glob("AIRR-*.json")): | |
| if entry.name.endswith(".review.json"): | |
| continue | |
| reports.append( | |
| { | |
| "report_id": entry.stem, | |
| "experiment_id": current_experiment, | |
| "json_path": str(entry.relative_to(self._root)).replace( | |
| "\\", "/" | |
| ), | |
| "markdown_path": str( | |
| entry.with_suffix(".md").relative_to(self._root) | |
| ).replace("\\", "/"), | |
| "size_bytes": entry.stat().st_size, | |
| } | |
| ) | |
| if experiment_id is not None: | |
| reports = [ | |
| report for report in reports if report["experiment_id"] == experiment_id | |
| ] | |
| return reports | |
| def experiment_manifest(self, experiment_id: str) -> dict[str, JSONValue] | None: | |
| """Load one experiment manifest.json if present.""" | |
| candidate = self._root / "experiments" / experiment_id / "manifest.json" | |
| if not candidate.is_file(): | |
| return None | |
| try: | |
| data: Any = json.loads(candidate.read_text(encoding="utf-8")) | |
| except (OSError, json.JSONDecodeError): | |
| return None | |
| return data # type: ignore[no-any-return] | |
| def _work_view_archived_ids(self) -> frozenset[str]: | |
| """Read the optional metadata-only archive index without mutating it.""" | |
| index = self._root / "archive" / "experiment_index.json" | |
| if not index.is_file(): | |
| return frozenset() | |
| try: | |
| payload: object = json.loads(index.read_text(encoding="utf-8")) | |
| except (OSError, json.JSONDecodeError): | |
| return frozenset() | |
| if not isinstance(payload, dict): | |
| return frozenset() | |
| experiments = cast(dict[object, object], payload).get("experiments", {}) | |
| if not isinstance(experiments, dict): | |
| return frozenset() | |
| return frozenset(str(key) for key in cast(dict[object, object], experiments)) | |
| def list_experiments(self) -> list[dict[str, JSONValue]]: | |
| """List experiments visible in the active work view. | |
| The archive index controls dashboard visibility for every experiment | |
| record kind, including historical campaign indexes. Archiving changes | |
| only this metadata view; canonical publication artifacts remain on disk. | |
| """ | |
| directory = self._root / "experiments" | |
| if not directory.is_dir(): | |
| return [] | |
| hidden = self._work_view_archived_ids() | |
| experiments: list[dict[str, JSONValue]] = [] | |
| for entry in directory.iterdir(): | |
| manifest = entry / "manifest.json" | |
| if not manifest.is_file(): | |
| continue | |
| try: | |
| data: Any = json.loads(manifest.read_text(encoding="utf-8")) | |
| except (OSError, json.JSONDecodeError): | |
| continue | |
| manifest_data = ( | |
| cast(dict[str, Any], data) if isinstance(data, dict) else None | |
| ) | |
| if entry.name in hidden: | |
| continue | |
| dashboard_manifest = ( | |
| normalize_experiment_manifest(entry, manifest_data) | |
| if manifest_data is not None | |
| else None | |
| ) | |
| created_at = ( | |
| str(dashboard_manifest.get("created_at")) | |
| if dashboard_manifest is not None | |
| and dashboard_manifest.get("created_at") | |
| else None | |
| ) | |
| experiments.append( | |
| { | |
| "ai_operation_mode": ( | |
| classify_ai_operation(manifest_data) | |
| if manifest_data is not None | |
| else "NONE" | |
| ), | |
| "research_operation_status": ( | |
| classify_research_operation_status(manifest_data) | |
| if manifest_data is not None | |
| else "UNKNOWN" | |
| ), | |
| "id": entry.name, | |
| "created_at": created_at, | |
| "path": str(entry.relative_to(self._root)).replace("\\", "/"), | |
| "manifest": ( | |
| dashboard_manifest if dashboard_manifest is not None else data | |
| ), | |
| } | |
| ) | |
| experiments.sort( | |
| key=lambda item: str(item.get("created_at") or ""), reverse=True | |
| ) | |
| return experiments | |
| def _safe_resolve(self, relative_path: str) -> Path | None: | |
| if not relative_path or ".." in relative_path: | |
| return None | |
| candidate = (self._root / relative_path).resolve() | |
| try: | |
| candidate.relative_to(self._root) | |
| except ValueError: | |
| return None | |
| if not candidate.is_file(): | |
| return None | |
| return candidate | |
| def create_research_source(research_root: Path | None) -> ResearchSource | None: | |
| """Create a ResearchSource if the root exists, else None.""" | |
| if research_root is None or not research_root.exists(): | |
| return None | |
| return ResearchSource(research_root) | |