BioScientist/agent_system/engines/e1_validator.py

from __future__ import annotations

import csv
import json
import os
import subprocess
import sys
import importlib.util
import statistics
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from ..schemas import Hypothesis, Insight, ValidationReport
from ..shared_memory import SharedKnowledgeSpace


class BioinfoE1Validator:
    """
    E1 validator: layered and cost-aware hypothesis validation.
    First version:
    - L1: logical consistency / rule checks
    - L2: light tooling readiness check using Biomni + registered MCP servers
    - L3: lightweight data-backed validation on biomni_data
    - L4: extended data-backed validation with fuller passes and numeric checks
    """

    LEVEL_CONFIDENCE = {
        "L1": 0.30,
        "L2": 0.60,
        "L3": 0.85,
        "L4": 0.95,
    }

    def __init__(
        self,
        memory: SharedKnowledgeSpace,
        biomni_root: str | Path,
        mcp_config_path: str | Path,
        project_root: str | Path,
    ):
        self.memory = memory
        self.biomni_root = Path(biomni_root).resolve()
        self.mcp_config_path = Path(mcp_config_path).resolve()
        self.project_root = Path(project_root).resolve()
        self._registration_state: dict[str, Any] | None = None
        self._biomni_probe_cache: dict[str, Any] | None = None
        self._biomni_agent_cache: dict[str, Any] | None = None

    def _resolve_register_script(self) -> Path | None:
        candidates = [
            # when project_root=/.../BioScientist/agent_system
            self.project_root / "toolbase" / "register_mcp_servers_to_biomni.py",
            # when project_root=/.../BioScientist
            self.project_root / "agent_system" / "toolbase" / "register_mcp_servers_to_biomni.py",
            # fallback based on current file location
            Path(__file__).resolve().parents[1] / "toolbase" / "register_mcp_servers_to_biomni.py",
        ]
        for c in candidates:
            if c.exists():
                return c
        return None

    def ensure_mcp_registered(self, dry_run: bool = False) -> dict[str, Any]:
        script = self._resolve_register_script()
        if script is None:
            return {
                "ok": False,
                "status": "missing_script",
                "error": (
                    "register script not found under either "
                    f"{self.project_root}/toolbase or {self.project_root}/agent_system/toolbase"
                ),
            }
        cmd = [
            sys.executable,
            str(script),
            "--biomni-root",
            str(self.biomni_root),
            "--config-out",
            str(self.mcp_config_path),
        ]
        if dry_run:
            cmd.append("--dry-run")
        try:
            completed = subprocess.run(cmd, capture_output=True, text=True, check=False)
            state = {
                "ok": completed.returncode == 0,
                "status": "registered" if completed.returncode == 0 else "failed",
                "return_code": completed.returncode,
                "stdout_tail": "\n".join((completed.stdout or "").splitlines()[-20:]),
                "stderr_tail": "\n".join((completed.stderr or "").splitlines()[-20:]),
                "config_path": str(self.mcp_config_path),
            }
            self._registration_state = state
            return state
        except Exception as exc:
            state = {"ok": False, "status": "failed", "error": str(exc), "config_path": str(self.mcp_config_path)}
            self._registration_state = state
            return state

    def _coerce_hypothesis(self, hypothesis: Hypothesis | dict[str, Any]) -> dict[str, Any]:
        return hypothesis.to_dict() if isinstance(hypothesis, Hypothesis) else hypothesis

    def _prepare_biomni_import(self) -> dict[str, Any]:
        """
        Ensure `import biomni` works without requiring pip install -e.
        """
        info = {
            "biomni_root": str(self.biomni_root),
            "path_injected": False,
            "package_dir_exists": (self.biomni_root / "biomni").exists(),
        }
        root_str = str(self.biomni_root)
        if root_str not in sys.path:
            sys.path.insert(0, root_str)
            info["path_injected"] = True
        return info

    def _resolve_biomni_data_root(self) -> Path:
        env = os.getenv("BIOCLAW_BIOMNI_DATA_ROOT", "").strip()
        if env:
            p = Path(env).expanduser().resolve()
            if p.exists():
                return p
        candidates = [
            self.project_root / "agent_system" / "toolbase" / "data" / "biomni_data",
            self.project_root / "toolbase" / "data" / "biomni_data",
            Path("/225040511/project/BioScientist/agent_system/toolbase/data/biomni_data"),
        ]
        for c in candidates:
            if c.exists():
                return c
        return candidates[0]

    def _resolve_biomni_llm_source(self) -> tuple[str, str]:
        """
        Resolve runtime LLM/source for Biomni execution.
        Priority:
        1) BIOCLAW_BIOMNI_LLM / BIOCLAW_BIOMNI_SOURCE
        2) GEMINI_MODEL / BIOMNI_SOURCE
        3) safe Gemini defaults
        """
        llm = (
            os.getenv("BIOCLAW_BIOMNI_LLM", "").strip()
            or os.getenv("GEMINI_MODEL", "").strip()
            or "gemini-2.5-flash-lite"
        )
        source = (
            os.getenv("BIOCLAW_BIOMNI_SOURCE", "").strip()
            or os.getenv("BIOMNI_SOURCE", "").strip()
            or "Gemini"
        )
        return llm, source

    def _quick_biomni_probe(self, hypothesis_text: str = "", force: bool = False) -> dict[str, Any]:
        """
        Minimal-cost Biomni runtime probe.
        Reuses cached probe; otherwise builds/uses a cached Biomni agent context.
        """
        if self._biomni_probe_cache is not None and not force:
            cached = dict(self._biomni_probe_cache)
            cached["cache_hit"] = True
            return cached

        t0 = datetime.now(timezone.utc)
        ctx = self._get_biomni_agent_context(hypothesis_text, force_rebuild=force)
        dt = datetime.now(timezone.utc) - t0
        if not ctx.get("ok", False):
            result = {
                "ok": False,
                "status": "biomni_probe_failed",
                "error": str(ctx.get("error", "unknown")),
                "mcp_registration_ok": bool(ctx.get("trace", {}).get("mcp_registration_ok", False)),
                "duration_ms": int(dt.total_seconds() * 1000),
                "cache_hit": False,
                "mcp_config_used": str(ctx.get("trace", {}).get("mcp_config_used", self.mcp_config_path)),
                **(ctx.get("trace", {}) if isinstance(ctx.get("trace", {}), dict) else {}),
            }
            self._biomni_probe_cache = result
            return result

        trace = ctx.get("trace", {}) if isinstance(ctx.get("trace", {}), dict) else {}
        result = {
            "ok": True,
            "status": "ready",
            "mcp_registration_ok": True,
            "custom_tool_count": int(trace.get("custom_tool_count", 0)),
            "related_tool_hits": 0,
            "duration_ms": int(dt.total_seconds() * 1000),
            "cache_hit": False,
            "mcp_config_used": str(trace.get("mcp_config_used", self.mcp_config_path)),
            "loaded_server_count": int(trace.get("loaded_server_count", -1)),
            "minimal_config_selected_count": int(trace.get("minimal_config_selected_count", 0)),
            "full_config_server_count": int(trace.get("full_config_server_count", 0)),
            "path_injected": bool(trace.get("path_injected", False)),
            "package_dir_exists": bool(trace.get("package_dir_exists", False)),
            "biomni_root": str(trace.get("biomni_root", self.biomni_root)),
        }
        self._biomni_probe_cache = result
        return result

    def _get_biomni_agent_context(self, hypothesis_text: str = "", force_rebuild: bool = False) -> dict[str, Any]:
        """
        Build/reuse a Biomni A1 agent with MCP loaded.
        Returns a context dict:
        {
          ok: bool,
          agent: A1 | None,
          trace: dict[str, Any],
          error: str
        }
        """
        if self._biomni_agent_cache is not None and not force_rebuild:
            return self._biomni_agent_cache

        registration = self._registration_state or self.ensure_mcp_registered(dry_run=False)
        if not registration.get("ok", False):
            ctx = {
                "ok": False,
                "agent": None,
                "trace": {"mcp_registration_ok": False},
                "error": "mcp_registration_failed",
            }
            self._biomni_agent_cache = ctx
            return ctx

        deps_ok, missing_modules, deps_evidence = self._check_l2_dependencies()
        if not deps_ok:
            ctx = {
                "ok": False,
                "agent": None,
                "trace": {
                    "mcp_registration_ok": True,
                    "missing_modules": missing_modules,
                },
                "error": deps_evidence,
            }
            self._biomni_agent_cache = ctx
            return ctx

        try:
            import_info = self._prepare_biomni_import()
            from biomni.agent import A1
            llm_name, llm_source = self._resolve_biomni_llm_source()

            min_cfg = self._build_minimal_mcp_config(hypothesis_text)
            used_config_path = str(self.mcp_config_path)
            loaded_server_count = -1
            minimal_server_count = 0
            full_server_count = 0
            if min_cfg is not None:
                minimal_server_count = int(min_cfg.get("selected_count", 0))
                full_server_count = int(min_cfg.get("full_count", 0))
                if minimal_server_count > 0:
                    used_config_path = str(min_cfg["path"])

            agent = A1(llm=llm_name, source=llm_source)
            agent.add_mcp(config_path=used_config_path)
            tools = agent.list_custom_tools() or []
            if min_cfg is not None and minimal_server_count > 0:
                loaded_server_count = minimal_server_count
            elif min_cfg is not None:
                loaded_server_count = full_server_count
            if not tools and used_config_path != str(self.mcp_config_path):
                agent = A1(llm=llm_name, source=llm_source)
                agent.add_mcp(config_path=str(self.mcp_config_path))
                tools = agent.list_custom_tools() or []
                used_config_path = str(self.mcp_config_path)
                loaded_server_count = full_server_count if full_server_count > 0 else -1

            ctx = {
                "ok": True,
                "agent": agent,
                "trace": {
                    "custom_tool_count": len(tools),
                    "mcp_config_used": used_config_path,
                    "loaded_server_count": loaded_server_count,
                    "minimal_config_selected_count": minimal_server_count,
                    "full_config_server_count": full_server_count,
                    "biomni_llm": llm_name,
                    "biomni_source": llm_source,
                    **import_info,
                },
                "error": "",
            }
            self._biomni_agent_cache = ctx
            return ctx
        except Exception as exc:
            ctx = {
                "ok": False,
                "agent": None,
                "trace": {"mcp_registration_ok": True},
                "error": str(exc),
            }
            self._biomni_agent_cache = ctx
            return ctx

    @staticmethod
    def _extract_json_like_block(text: str) -> dict[str, Any]:
        s = (text or "").strip()
        if not s:
            return {}
        try:
            return json.loads(s) if s.startswith("{") else {}
        except Exception:
            pass
        # naive fallback: find first {...}
        start = s.find("{")
        end = s.rfind("}")
        if start >= 0 and end > start:
            block = s[start : end + 1]
            try:
                return json.loads(block)
            except Exception:
                return {}
        return {}

    def _execute_hypothesis_with_biomni(self, h: dict[str, Any]) -> dict[str, Any]:
        """
        Execute hypothesis using Biomni directly (A1.go) with lightweight prompt.
        Returns execution summary and artifact path.
        """
        text = f"{h.get('title', '')} {h.get('hypothesis', '')}".strip()
        ctx = self._get_biomni_agent_context(text)
        if not ctx.get("ok", False):
            return {
                "ok": False,
                "error": f"biomni_agent_unavailable: {ctx.get('error', 'unknown')}",
                "trace": ctx.get("trace", {}),
            }

        agent = ctx.get("agent")
        data_sources = [str(x) for x in h.get("data_sources", [])][:5]
        ops = [x for x in h.get("data_operations", []) if isinstance(x, dict)]
        prompt = (
            "You are validating a bioinformatics hypothesis in lightweight mode.\n"
            "Use available MCP tools where useful and provide concise execution output.\n"
            "Return JSON with keys: status, summary, key_metrics, recommendation.\n\n"
            f"Domain: {h.get('domain', '')}\n"
            f"Hypothesis title: {h.get('title', '')}\n"
            f"Hypothesis: {h.get('hypothesis', '')}\n"
            f"Expected improvement: {h.get('expected_improvement', '')}\n"
            f"Theoretical basis: {h.get('theoretical_basis', '')}\n"
            f"Data sources: {json.dumps(data_sources, ensure_ascii=True)}\n"
            f"Requested operations: {json.dumps(ops, ensure_ascii=True)}\n"
            "Please execute a minimal-cost validation and return machine-readable JSON."
        )

        try:
            logs, final_message = agent.go(prompt)
            parsed = self._extract_json_like_block(str(final_message))
            status_raw = str(parsed.get("status", "")).lower()
            mapped_status = "success" if status_raw in {"success", "ok", "passed"} else ("failed" if status_raw in {"failed", "error"} else "inconclusive")
            summary = str(parsed.get("summary", "")).strip() or str(final_message)[:1000]
            key_metrics = parsed.get("key_metrics", {}) if isinstance(parsed.get("key_metrics", {}), dict) else {}
            recommendation = str(parsed.get("recommendation", "")).strip()

            out_dir = self.project_root / "agent_system" / "results" / "biomni_exec"
            out_dir.mkdir(parents=True, exist_ok=True)
            ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
            out_path = out_dir / f"biomni_exec_{h.get('hypothesis_id', 'unknown')}_{ts}.json"
            out_payload = {
                "hypothesis_id": h.get("hypothesis_id", ""),
                "domain": h.get("domain", ""),
                "prompt": prompt,
                "parsed": parsed,
                "status": mapped_status,
                "summary": summary,
                "key_metrics": key_metrics,
                "recommendation": recommendation,
                "final_message": str(final_message),
                "log_tail": logs[-20:] if isinstance(logs, list) else [],
                "trace": ctx.get("trace", {}),
                "saved_at": datetime.now(timezone.utc).isoformat(),
            }
            out_path.write_text(json.dumps(out_payload, indent=2, ensure_ascii=True), encoding="utf-8")
            return {
                "ok": True,
                "status": mapped_status,
                "summary": summary,
                "key_metrics": key_metrics,
                "recommendation": recommendation,
                "artifact_file": str(out_path),
                "trace": ctx.get("trace", {}),
            }
        except Exception as exc:
            return {
                "ok": False,
                "error": f"biomni_execution_failed: {exc}",
                "trace": ctx.get("trace", {}),
            }

    def _build_minimal_mcp_config(self, hypothesis_text: str) -> dict[str, Any] | None:
        yaml_spec = importlib.util.find_spec("yaml")
        if yaml_spec is None:
            return None
        import yaml  # type: ignore

        if not self.mcp_config_path.exists():
            return None

        try:
            raw = self.mcp_config_path.read_text(encoding="utf-8")
            doc = yaml.safe_load(raw) or {}
        except Exception:
            return None
        servers = doc.get("mcp_servers", {})
        if not isinstance(servers, dict) or not servers:
            return None

        full_count = len(servers)
        max_servers = int(os.getenv("BIOCLAW_MIN_MCP_SERVERS", "40"))
        max_servers = max(5, min(200, max_servers))

        q = (hypothesis_text or "").lower()
        tokens = [t for t in q.replace("-", " ").replace("_", " ").split() if len(t) >= 3][:30]

        scored: list[tuple[int, str, dict[str, Any]]] = []
        for name, cfg in servers.items():
            if not isinstance(cfg, dict):
                continue
            cmd = " ".join([str(x).lower() for x in cfg.get("command", [])]) if isinstance(cfg.get("command"), list) else str(cfg.get("command", "")).lower()
            desc = str(cfg.get("description", "")).lower()
            hay = f"{name.lower()} {desc} {cmd}"
            s = 0
            for tok in tokens:
                if tok in hay:
                    s += 2
            if "single" in hay or "cell" in hay or "rna" in hay:
                s += 1
            scored.append((s, name, cfg))

        scored.sort(key=lambda x: x[0], reverse=True)
        chosen = [x for x in scored if x[0] > 0][:max_servers]
        if len(chosen) < min(10, max_servers):
            # Backfill with top entries to keep probe meaningful.
            fallback_pool = [x for x in scored if x[1] not in {c[1] for c in chosen}]
            need = min(10, max_servers) - len(chosen)
            chosen.extend(fallback_pool[:need])
        if not chosen:
            chosen = scored[: min(10, max_servers)]
        if not chosen:
            return None

        selected_servers = {name: cfg for _, name, cfg in chosen}
        out_dir = self.project_root / "agent_system" / "results" / "e1_runtime" / "min_configs"
        out_dir.mkdir(parents=True, exist_ok=True)
        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        out_path = out_dir / f"mcp_min_{ts}.yaml"
        out_path.write_text(
            yaml.safe_dump({"mcp_servers": selected_servers}, sort_keys=False, allow_unicode=False),
            encoding="utf-8",
        )
        return {
            "path": out_path,
            "selected_count": len(selected_servers),
            "full_count": full_count,
        }

    def _persist_runtime_artifact(self, report_dict: dict[str, Any], hypothesis: dict[str, Any]) -> str:
        runtime_dir = self.project_root / "agent_system" / "results" / "e1_runtime"
        runtime_dir.mkdir(parents=True, exist_ok=True)
        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        out_path = runtime_dir / f"{report_dict.get('level', 'Lx').lower()}_{report_dict.get('validation_id', 'unknown')}_{ts}.json"
        payload = {
            "validation": report_dict,
            "hypothesis": {
                "hypothesis_id": hypothesis.get("hypothesis_id", ""),
                "domain": hypothesis.get("domain", ""),
                "title": hypothesis.get("title", ""),
            },
            "saved_at": datetime.now(timezone.utc).isoformat(),
        }
        out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=True), encoding="utf-8")

        summary_path = runtime_dir / "summary.json"
        if summary_path.exists():
            try:
                summary = json.loads(summary_path.read_text(encoding="utf-8"))
            except Exception:
                summary = {}
        else:
            summary = {}
        by_level = summary.setdefault("by_level", {})
        lvl = report_dict.get("level", "UNKNOWN")
        s = by_level.setdefault(lvl, {"count": 0, "success": 0, "inconclusive": 0, "failed": 0, "avg_score": 0.0})
        old_count = int(s.get("count", 0))
        old_avg = float(s.get("avg_score", 0.0))
        new_score = float(report_dict.get("score", 0.0))
        new_count = old_count + 1
        s["count"] = new_count
        s["avg_score"] = (old_avg * old_count + new_score) / max(1, new_count)
        st = str(report_dict.get("status", "inconclusive"))
        s[st] = int(s.get(st, 0)) + 1
        summary["total_reports"] = int(summary.get("total_reports", 0)) + 1
        summary["last_report_file"] = str(out_path)
        summary["last_updated"] = datetime.now(timezone.utc).isoformat()
        summary_path.write_text(json.dumps(summary, indent=2, ensure_ascii=True), encoding="utf-8")
        return str(out_path)

    def _index_data_files(self, data_root: Path, limit: int = 1000) -> list[Path]:
        if not data_root.exists():
            return []
        files: list[Path] = []
        for p in sorted(data_root.rglob("*")):
            if p.is_file():
                files.append(p)
                if len(files) >= limit:
                    break
        return files

    def _download_online_example_data(self, domain: str, max_files: int = 2) -> list[Path]:
        """
        Best-effort fallback downloader when no usable local dataset is available.
        Downloads small public tabular files so L3/L4 can still produce bound validation reports.
        """
        out_dir = self.project_root / "agent_system" / "results" / "downloaded_example_data"
        out_dir.mkdir(parents=True, exist_ok=True)

        domain_l = str(domain).lower()
        single_cell_candidates = [
            (
                "pbmc3k_marker_genes.csv",
                "https://raw.githubusercontent.com/scverse/scanpy-tutorials/main/pbmc3k/marker_genes.csv",
            ),
            (
                "pbmc3k_obs_metadata.csv",
                "https://raw.githubusercontent.com/scverse/scanpy-tutorials/main/pbmc3k/obs_metadata.csv",
            ),
        ]
        generic_candidates = [
            (
                "iris.csv",
                "https://raw.githubusercontent.com/mwaskom/seaborn-data/master/iris.csv",
            ),
            (
                "tips.csv",
                "https://raw.githubusercontent.com/mwaskom/seaborn-data/master/tips.csv",
            ),
        ]
        candidates = single_cell_candidates + generic_candidates if ("single" in domain_l or "cell" in domain_l) else generic_candidates + single_cell_candidates

        downloaded: list[Path] = []
        for filename, url in candidates:
            target = out_dir / filename
            if target.exists() and target.stat().st_size > 0:
                downloaded.append(target)
                if len(downloaded) >= max_files:
                    break
                continue
            try:
                urllib.request.urlretrieve(url, str(target))
                if target.exists() and target.stat().st_size > 0:
                    downloaded.append(target)
                if len(downloaded) >= max_files:
                    break
            except Exception:
                continue
        return downloaded

    def _safe_read_tabular_preview(self, file_path: Path, sample_rows: int = 500) -> dict[str, Any]:
        suffix = file_path.suffix.lower()
        result: dict[str, Any] = {
            "file": str(file_path),
            "suffix": suffix,
            "status": "skipped",
            "rows_scanned": 0,
            "columns": [],
            "missing_cells": 0,
            "total_cells": 0,
            "numeric_columns": [],
        }

        if suffix in {".csv", ".tsv"}:
            delim = "," if suffix == ".csv" else "\t"
            with file_path.open("r", encoding="utf-8", errors="ignore", newline="") as fh:
                reader = csv.DictReader(fh, delimiter=delim)
                cols = reader.fieldnames or []
                result["columns"] = cols
                rows = 0
                missing = 0
                total = 0
                numeric_candidates: dict[str, int] = {c: 0 for c in cols}
                for row in reader:
                    rows += 1
                    for c in cols:
                        v = (row.get(c) or "").strip()
                        total += 1
                        if v == "":
                            missing += 1
                        else:
                            try:
                                float(v)
                                numeric_candidates[c] += 1
                            except Exception:
                                pass
                    if rows >= sample_rows:
                        break
                result["rows_scanned"] = rows
                result["missing_cells"] = missing
                result["total_cells"] = total
                result["numeric_columns"] = [c for c, n in numeric_candidates.items() if n > 0]
                result["status"] = "loaded"
                return result

        if suffix == ".parquet":
            pd_spec = importlib.util.find_spec("pandas")
            if pd_spec is None:
                result["status"] = "skipped"
                result["error"] = "pandas_not_installed_for_parquet"
                return result
            import pandas as pd  # type: ignore

            df = pd.read_parquet(file_path)
            if len(df) > sample_rows:
                df = df.head(sample_rows)
            result["rows_scanned"] = int(len(df))
            cols = [str(c) for c in df.columns.tolist()]
            result["columns"] = cols
            total = int(df.shape[0] * df.shape[1]) if df.shape[1] > 0 else 0
            missing = int(df.isna().sum().sum()) if total > 0 else 0
            numeric_cols = [str(c) for c in df.select_dtypes(include=["number"]).columns.tolist()]
            result["missing_cells"] = missing
            result["total_cells"] = total
            result["numeric_columns"] = numeric_cols
            result["status"] = "loaded"
            return result

        if suffix == ".json":
            with file_path.open("r", encoding="utf-8", errors="ignore") as fh:
                payload = json.load(fh)
            if isinstance(payload, list):
                rows = min(sample_rows, len(payload))
                cols = sorted({k for item in payload[:rows] if isinstance(item, dict) for k in item.keys()})
                result["rows_scanned"] = rows
                result["columns"] = cols
                result["status"] = "loaded"
            elif isinstance(payload, dict):
                result["rows_scanned"] = 1
                result["columns"] = sorted(payload.keys())
                result["status"] = "loaded"
            else:
                result["status"] = "loaded"
            return result

        result["status"] = "skipped"
        result["error"] = f"unsupported_suffix:{suffix or 'none'}"
        return result

    def _safe_read_tabular_full(self, file_path: Path, max_rows: int = 200000) -> dict[str, Any]:
        """
        L4-oriented data pass.
        Attempts a fuller scan (or up to max_rows for safety) and includes basic numeric summaries.
        """
        suffix = file_path.suffix.lower()
        result: dict[str, Any] = {
            "file": str(file_path),
            "suffix": suffix,
            "status": "skipped",
            "rows_scanned": 0,
            "columns": [],
            "missing_cells": 0,
            "total_cells": 0,
            "numeric_columns": [],
            "truncated": False,
            "numeric_summary": {},
        }

        if suffix in {".csv", ".tsv"}:
            delim = "," if suffix == ".csv" else "\t"
            with file_path.open("r", encoding="utf-8", errors="ignore", newline="") as fh:
                reader = csv.DictReader(fh, delimiter=delim)
                cols = reader.fieldnames or []
                result["columns"] = cols
                rows = 0
                missing = 0
                total = 0
                numeric_samples: dict[str, list[float]] = {c: [] for c in cols}
                for row in reader:
                    rows += 1
                    for c in cols:
                        v = (row.get(c) or "").strip()
                        total += 1
                        if v == "":
                            missing += 1
                        else:
                            try:
                                fv = float(v)
                                if len(numeric_samples[c]) < 5000:
                                    numeric_samples[c].append(fv)
                            except Exception:
                                pass
                    if rows >= max_rows:
                        result["truncated"] = True
                        break
                result["rows_scanned"] = rows
                result["missing_cells"] = missing
                result["total_cells"] = total
                numeric_cols = [c for c, arr in numeric_samples.items() if arr]
                result["numeric_columns"] = numeric_cols
                summary = {}
                for c in numeric_cols[:20]:
                    arr = numeric_samples[c]
                    summary[c] = {
                        "count": len(arr),
                        "mean": float(statistics.fmean(arr)),
                        "min": float(min(arr)),
                        "max": float(max(arr)),
                    }
                result["numeric_summary"] = summary
                result["status"] = "loaded"
                return result

        if suffix == ".parquet":
            pd_spec = importlib.util.find_spec("pandas")
            if pd_spec is None:
                result["status"] = "skipped"
                result["error"] = "pandas_not_installed_for_parquet"
                return result
            import pandas as pd  # type: ignore

            df = pd.read_parquet(file_path)
            if len(df) > max_rows:
                df = df.head(max_rows)
                result["truncated"] = True
            result["rows_scanned"] = int(len(df))
            cols = [str(c) for c in df.columns.tolist()]
            result["columns"] = cols
            total = int(df.shape[0] * df.shape[1]) if df.shape[1] > 0 else 0
            missing = int(df.isna().sum().sum()) if total > 0 else 0
            numeric_cols = [str(c) for c in df.select_dtypes(include=["number"]).columns.tolist()]
            result["missing_cells"] = missing
            result["total_cells"] = total
            result["numeric_columns"] = numeric_cols
            summary = {}
            for c in numeric_cols[:20]:
                s = df[c].dropna()
                if len(s) == 0:
                    continue
                summary[str(c)] = {
                    "count": int(len(s)),
                    "mean": float(s.mean()),
                    "min": float(s.min()),
                    "max": float(s.max()),
                }
            result["numeric_summary"] = summary
            result["status"] = "loaded"
            return result

        if suffix == ".json":
            with file_path.open("r", encoding="utf-8", errors="ignore") as fh:
                payload = json.load(fh)
            if isinstance(payload, list):
                rows = len(payload)
                if rows > max_rows:
                    rows = max_rows
                    result["truncated"] = True
                cols = sorted({k for item in payload[:rows] if isinstance(item, dict) for k in item.keys()})
                result["rows_scanned"] = rows
                result["columns"] = cols
                result["status"] = "loaded"
            elif isinstance(payload, dict):
                result["rows_scanned"] = 1
                result["columns"] = sorted(payload.keys())
                result["status"] = "loaded"
            else:
                result["status"] = "loaded"
            return result

        result["status"] = "skipped"
        result["error"] = f"unsupported_suffix:{suffix or 'none'}"
        return result

    @staticmethod
    def _domain_relevance_score(domain: str, selected_files: list[Path], per_file_results: list[dict[str, Any]]) -> tuple[float, list[str]]:
        domain_l = str(domain).lower()
        if not selected_files:
            return 0.0, ["no_selected_files"]
        hints = []
        if "single_cell" in domain_l or "single" in domain_l:
            hints = ["single", "cell", "rna", "census", "marker", "celltype", "expression"]
        else:
            hints = [x for x in domain_l.split("_") if x]

        matched = 0
        explanations: list[str] = []
        for p in selected_files:
            p_low = str(p).lower()
            if any(h in p_low for h in hints):
                matched += 1
                explanations.append(f"file_match:{p.name}")
        for r in per_file_results:
            cols = " ".join([str(c).lower() for c in r.get("columns", [])])
            if any(h in cols for h in hints):
                matched += 1
                explanations.append(f"column_match:{Path(str(r.get('file', 'unknown'))).name}")
        denom = max(1, len(selected_files) + len(per_file_results))
        return min(1.0, matched / denom), explanations[:10]

    @staticmethod
    def _build_interpretability_summary(
        *,
        level: str,
        loaded_count: int,
        selected_count: int,
        rows_scanned: int,
        missing_rate: float,
        numeric_cols_total: int,
        relevance_score: float,
    ) -> dict[str, Any]:
        quality = "high" if missing_rate < 0.01 else ("medium" if missing_rate < 0.1 else "low")
        coverage = loaded_count / max(1, selected_count)
        coverage_label = "high" if coverage >= 0.8 else ("medium" if coverage >= 0.5 else "low")
        relevance_label = "high" if relevance_score >= 0.6 else ("medium" if relevance_score >= 0.3 else "low")
        return {
            "level": level,
            "data_quality": quality,
            "coverage": coverage_label,
            "domain_relevance": relevance_label,
            "rows_scanned": rows_scanned,
            "numeric_signal_strength": "high" if numeric_cols_total >= 10 else ("medium" if numeric_cols_total >= 3 else "low"),
            "recommendations": [
                "Prefer domain-relevant files if domain_relevance is low.",
                "Increase file coverage when coverage is medium/low.",
                "Escalate to L4 only after L3 relevance is acceptable.",
            ],
        }

    @staticmethod
    def _check_l2_dependencies() -> tuple[bool, list[str], str]:
        required_modules = ["nest_asyncio", "mcp", "yaml"]
        missing = [m for m in required_modules if importlib.util.find_spec(m) is None]
        if not missing:
            return True, [], ""

        module_to_pip = {
            "nest_asyncio": "nest_asyncio",
            "mcp": "mcp",
            "yaml": "PyYAML",
        }
        pip_pkgs = [module_to_pip[m] for m in missing]
        install_cmd = f"{sys.executable} -m pip install " + " ".join(pip_pkgs)
        evidence = (
            "L2 dependency check failed. Missing modules: "
            f"{', '.join(missing)}. "
            f"Install with: {install_cmd}. "
            f"(module->pip: {', '.join(f'{m}->{module_to_pip[m]}' for m in missing)})"
        )
        return False, missing, evidence

    def _validate_l1(self, h: dict[str, Any]) -> tuple[str, float, dict[str, Any], str, str]:
        text = f"{h.get('title', '')} {h.get('hypothesis', '')}".lower()
        violations: list[str] = []
        if "敲除" in text and "增殖加快" in text:
            violations.append("possible essential-gene-growth contradiction")
        if "不存在的工具" in text:
            violations.append("depends on non-existing tool")
        if "必须" in text and "不需要数据" in text:
            violations.append("self-contradictory requirement")

        if violations:
            return (
                "failed",
                0.1,
                {"rule_violations": violations, "violation_count": len(violations)},
                "biological_or_tooling_inconsistency",
                "L1 rules detected contradictions.",
            )
        return (
            "success",
            0.6,
            {"rule_violations": [], "violation_count": 0},
            "",
            "L1 rules found no obvious contradiction.",
        )

    def _validate_l2(self, h: dict[str, Any]) -> tuple[str, float, dict[str, Any], str, str]:
        registration = self._registration_state or self.ensure_mcp_registered(dry_run=False)
        if not registration.get("ok", False):
            return (
                "inconclusive",
                0.3,
                {"mcp_registration_ok": False},
                "mcp_registration_failed",
                "L2 could not ensure MCP registration.",
            )

        deps_ok, missing_modules, deps_evidence = self._check_l2_dependencies()
        if not deps_ok:
            return (
                "inconclusive",
                0.2,
                {
                    "mcp_registration_ok": True,
                    "l2_dependency_check_ok": False,
                    "missing_modules": missing_modules,
                },
                "biomni_dependency_missing",
                deps_evidence,
            )

        try:
            import_info = self._prepare_biomni_import()
            from biomni.agent import A1
            llm_name, llm_source = self._resolve_biomni_llm_source()

            agent = A1(llm=llm_name, source=llm_source)
            agent.add_mcp(config_path=str(self.mcp_config_path))
            tools = agent.list_custom_tools() or []
            if not tools:
                return (
                    "inconclusive",
                    0.35,
                    {
                        "mcp_registration_ok": True,
                        "custom_tool_count": 0,
                        "biomni_llm": llm_name,
                        "biomni_source": llm_source,
                        **import_info,
                    },
                    "no_mcp_tools_discovered",
                    "Biomni loaded config but no tools were discovered.",
                )

            text = f"{h.get('title', '')} {h.get('hypothesis', '')}".lower()
            related = [t for t in tools if any(tok in t.lower() for tok in text.split()[:20])]
            score = 0.5 + min(0.4, len(related) * 0.05)
            return (
                "success",
                score,
                {
                    "mcp_registration_ok": True,
                    "custom_tool_count": len(tools),
                    "related_tool_hits": len(related),
                    "biomni_llm": llm_name,
                    "biomni_source": llm_source,
                    **import_info,
                },
                "",
                "L2 passed using Biomni MCP tool readiness check.",
            )
        except Exception as exc:
            import_info = self._prepare_biomni_import()
            return (
                "inconclusive",
                0.25,
                {"mcp_registration_ok": True, **import_info},
                "biomni_probe_failed",
                f"Biomni probe failed: {exc}",
            )

    def _validate_l3(self, h: dict[str, Any]) -> tuple[str, float, dict[str, Any], str, str]:
        probe = self._quick_biomni_probe(f"{h.get('title', '')} {h.get('hypothesis', '')}")
        if not probe.get("ok", False):
            return (
                "inconclusive",
                0.25,
                {
                    "mcp_registration_ok": bool(probe.get("mcp_registration_ok", False)),
                    "biomni_probe_status": probe.get("status", ""),
                    "biomni_probe_duration_ms": probe.get("duration_ms", 0),
                    "biomni_call_trace": {
                        "registration_attempted": True,
                        "biomni_runtime_called": False,
                        "cache_hit": bool(probe.get("cache_hit", False)),
                    },
                },
                str(probe.get("status", "biomni_probe_failed")),
                f"L3 aborted: Biomni quick probe failed ({probe.get('error', 'unknown_error')}).",
            )

        data_root = self._resolve_biomni_data_root()
        data_files = self._index_data_files(data_root, limit=1500)
        if not data_files:
            return (
                "inconclusive",
                0.3,
                {"data_root": str(data_root), "indexed_files": 0},
                "l3_data_unavailable",
                "L3 could not locate biomni_data files.",
            )

        requested = [Path(str(x)) for x in h.get("data_sources", []) if str(x).strip()]
        if requested:
            selected = [p for p in requested if p.exists()][:5]
        else:
            selected = data_files[:5]
        downloaded_examples: list[Path] = []
        if not selected:
            downloaded_examples = self._download_online_example_data(h.get("domain", ""), max_files=2)
            selected = downloaded_examples[:]

        operations = [x for x in h.get("data_operations", []) if isinstance(x, dict)]
        if not operations:
            operations = [
                {"action": "load_table", "target": "all_selected"},
                {"action": "sample_rows", "rows": 500},
                {"action": "profile_missingness"},
                {"action": "numeric_summary"},
            ]

        per_file_results: list[dict[str, Any]] = []
        loaded_count = 0
        for p in selected:
            try:
                r = self._safe_read_tabular_preview(p, sample_rows=500)
                per_file_results.append(r)
                if r.get("status") == "loaded":
                    loaded_count += 1
            except Exception as exc:
                per_file_results.append(
                    {
                        "file": str(p),
                        "status": "failed",
                        "error": str(exc),
                    }
                )

        op_names = [str(op.get("action", "")) for op in operations]
        rows_scanned = sum(int(r.get("rows_scanned", 0) or 0) for r in per_file_results)
        total_cells = sum(int(r.get("total_cells", 0) or 0) for r in per_file_results)
        missing_cells = sum(int(r.get("missing_cells", 0) or 0) for r in per_file_results)
        missing_rate = (missing_cells / total_cells) if total_cells > 0 else 0.0
        numeric_cols_total = sum(len(r.get("numeric_columns", []) or []) for r in per_file_results)
        relevance_score, relevance_evidence = self._domain_relevance_score(h.get("domain", ""), selected, per_file_results)

        results_dir = self.project_root / "agent_system" / "results" / "l3_reports"
        results_dir.mkdir(parents=True, exist_ok=True)
        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        artifact_path = results_dir / f"l3_{h.get('hypothesis_id', 'unknown')}_{ts}.json"
        artifact_payload = {
            "hypothesis_id": h.get("hypothesis_id", ""),
            "domain": h.get("domain", ""),
            "data_root": str(data_root),
            "selected_files": [str(p) for p in selected],
            "operations": operations,
            "per_file_results": per_file_results,
            "rows_scanned": rows_scanned,
            "missing_rate": missing_rate,
        }
        artifact_path.write_text(json.dumps(artifact_payload, indent=2, ensure_ascii=True), encoding="utf-8")

        if loaded_count == 0:
            return (
                "inconclusive",
                0.35,
                {
                    "data_root": str(data_root),
                    "indexed_files": len(data_files),
                    "selected_files": [str(p) for p in selected],
                    "loaded_files": 0,
                    "operations_requested": op_names,
                    "l3_report_file": str(artifact_path),
                },
                "l3_no_loadable_data",
                "L3 selected files but none were loadable in lightweight mode.",
            )

        score_components = {
            "base": 0.45,
            "loaded_files_component": 0.08 * loaded_count,
            "operation_component": 0.03 * min(8, len(op_names)),
            "domain_relevance_component": 0.08 * relevance_score,
        }
        score = min(0.92, sum(score_components.values()))
        status = "success" if loaded_count >= 1 else "inconclusive"
        biomni_exec = self._execute_hypothesis_with_biomni(h)
        if biomni_exec.get("ok", False):
            b_status = biomni_exec.get("status", "inconclusive")
            score = min(1.0, score + (0.08 if b_status == "success" else 0.0))
            if b_status == "failed":
                status = "failed"
            elif status != "failed" and b_status == "inconclusive":
                status = "inconclusive"
        else:
            score = max(0.0, score - 0.08)
            if status == "success":
                status = "inconclusive"
        evidence = (
            "L3 executed lightweight full-data checks with sampled previews; "
            f"loaded {loaded_count}/{len(selected)} files, scanned_rows={rows_scanned}, "
            f"missing_rate={missing_rate:.4f}, domain_relevance={relevance_score:.2f}. Report: {artifact_path}"
        )
        return (
            status,
            score,
            {
                "data_root": str(data_root),
                "indexed_files": len(data_files),
                "selected_files": [str(p) for p in selected],
                "loaded_files": loaded_count,
                "rows_scanned": rows_scanned,
                "missing_rate": missing_rate,
                "numeric_columns_total": numeric_cols_total,
                "operations_requested": op_names,
                "domain_relevance_score": relevance_score,
                "domain_relevance_evidence": relevance_evidence,
                "score_breakdown": score_components,
                "interpretability_summary": self._build_interpretability_summary(
                    level="L3",
                    loaded_count=loaded_count,
                    selected_count=len(selected),
                    rows_scanned=rows_scanned,
                    missing_rate=missing_rate,
                    numeric_cols_total=numeric_cols_total,
                    relevance_score=relevance_score,
                ),
                "biomni_call_trace": {
                    "registration_attempted": bool(self._registration_state),
                    "biomni_runtime_called": True,
                    "cache_hit": bool(probe.get("cache_hit", False)),
                    "biomni_probe_duration_ms": probe.get("duration_ms", 0),
                    "custom_tool_count": probe.get("custom_tool_count", 0),
                    "related_tool_hits": probe.get("related_tool_hits", 0),
                    "mcp_config_used": probe.get("mcp_config_used", str(self.mcp_config_path)),
                    "loaded_server_count": probe.get("loaded_server_count", -1),
                    "minimal_config_selected_count": probe.get("minimal_config_selected_count", 0),
                    "full_config_server_count": probe.get("full_config_server_count", 0),
                    "note": "L3 first performs a minimal Biomni runtime probe, then runs lightweight data checks.",
                },
                "downloaded_example_files": [str(p) for p in downloaded_examples],
                "l3_report_file": str(artifact_path),
                "biomni_execution": biomni_exec,
            },
            "" if biomni_exec.get("ok", False) else "biomni_execution_failed",
            evidence,
        )

    def _validate_l4(self, h: dict[str, Any]) -> tuple[str, float, dict[str, Any], str, str]:
        probe = self._quick_biomni_probe(f"{h.get('title', '')} {h.get('hypothesis', '')}")
        if not probe.get("ok", False):
            return (
                "inconclusive",
                0.3,
                {
                    "mcp_registration_ok": bool(probe.get("mcp_registration_ok", False)),
                    "biomni_probe_status": probe.get("status", ""),
                    "biomni_probe_duration_ms": probe.get("duration_ms", 0),
                    "biomni_call_trace": {
                        "registration_attempted": True,
                        "biomni_runtime_called": False,
                        "cache_hit": bool(probe.get("cache_hit", False)),
                    },
                },
                str(probe.get("status", "biomni_probe_failed")),
                f"L4 aborted: Biomni quick probe failed ({probe.get('error', 'unknown_error')}).",
            )

        data_root = self._resolve_biomni_data_root()
        data_files = self._index_data_files(data_root, limit=3000)
        if not data_files:
            return (
                "inconclusive",
                0.3,
                {"data_root": str(data_root), "indexed_files": 0},
                "l4_data_unavailable",
                "L4 could not locate biomni_data files.",
            )

        requested = [Path(str(x)) for x in h.get("data_sources", []) if str(x).strip()]
        if requested:
            selected = [p for p in requested if p.exists()][:8]
        else:
            selected = data_files[:5]
        downloaded_examples: list[Path] = []
        if not selected:
            downloaded_examples = self._download_online_example_data(h.get("domain", ""), max_files=3)
            selected = downloaded_examples[:]

        operations = [x for x in h.get("data_operations", []) if isinstance(x, dict)]
        if not operations:
            operations = [
                {"action": "load_table", "target": "all_selected"},
                {"action": "sample_rows", "rows": 500},
                {"action": "profile_missingness"},
                {"action": "numeric_summary"},
            ]

        max_rows = int(os.getenv("BIOCLAW_L4_MAX_ROWS", "200000"))
        max_rows = max(1000, min(1000000, max_rows))

        per_file_results: list[dict[str, Any]] = []
        loaded_count = 0
        truncated_count = 0
        for p in selected:
            try:
                r = self._safe_read_tabular_full(p, max_rows=max_rows)
                per_file_results.append(r)
                if r.get("status") == "loaded":
                    loaded_count += 1
                if r.get("truncated"):
                    truncated_count += 1
            except Exception as exc:
                per_file_results.append(
                    {
                        "file": str(p),
                        "status": "failed",
                        "error": str(exc),
                    }
                )

        rows_scanned = sum(int(r.get("rows_scanned", 0) or 0) for r in per_file_results)
        total_cells = sum(int(r.get("total_cells", 0) or 0) for r in per_file_results)
        missing_cells = sum(int(r.get("missing_cells", 0) or 0) for r in per_file_results)
        missing_rate = (missing_cells / total_cells) if total_cells > 0 else 0.0
        numeric_cols_total = sum(len(r.get("numeric_columns", []) or []) for r in per_file_results)
        relevance_score, relevance_evidence = self._domain_relevance_score(h.get("domain", ""), selected, per_file_results)
        op_names = [str(op.get("action", "")) for op in operations]

        results_dir = self.project_root / "agent_system" / "results" / "l4_reports"
        results_dir.mkdir(parents=True, exist_ok=True)
        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        artifact_path = results_dir / f"l4_{h.get('hypothesis_id', 'unknown')}_{ts}.json"
        artifact_payload = {
            "hypothesis_id": h.get("hypothesis_id", ""),
            "domain": h.get("domain", ""),
            "data_root": str(data_root),
            "selected_files": [str(p) for p in selected],
            "operations": operations,
            "max_rows_per_file": max_rows,
            "per_file_results": per_file_results,
            "rows_scanned": rows_scanned,
            "missing_rate": missing_rate,
            "numeric_columns_total": numeric_cols_total,
            "truncated_file_count": truncated_count,
        }
        artifact_path.write_text(json.dumps(artifact_payload, indent=2, ensure_ascii=True), encoding="utf-8")

        if loaded_count == 0:
            return (
                "inconclusive",
                0.35,
                {
                    "data_root": str(data_root),
                    "indexed_files": len(data_files),
                    "selected_files": [str(p) for p in selected],
                    "loaded_files": 0,
                    "l4_report_file": str(artifact_path),
                },
                "l4_no_loadable_data",
                "L4 selected files but none were loadable.",
            )

        completeness = 1.0 - min(1.0, missing_rate * 10.0)
        coverage = min(1.0, loaded_count / max(1, len(selected)))
        richness = min(1.0, numeric_cols_total / 20.0)
        trunc_penalty = 0.08 if truncated_count > 0 else 0.0
        score_components = {
            "base": 0.45,
            "coverage_component": 0.25 * coverage,
            "completeness_component": 0.2 * completeness,
            "richness_component": 0.15 * richness,
            "domain_relevance_component": 0.1 * relevance_score,
            "truncation_penalty": -trunc_penalty,
        }
        score = max(0.0, min(1.0, sum(score_components.values())))
        status = "success" if (loaded_count >= max(1, min(2, len(selected))) and completeness >= 0.5) else "inconclusive"
        biomni_exec = self._execute_hypothesis_with_biomni(h)
        if biomni_exec.get("ok", False):
            b_status = biomni_exec.get("status", "inconclusive")
            score = min(1.0, score + (0.1 if b_status == "success" else 0.0))
            if b_status == "failed":
                status = "failed"
            elif status != "failed" and b_status == "inconclusive":
                status = "inconclusive"
        else:
            score = max(0.0, score - 0.1)
            if status == "success":
                status = "inconclusive"
        evidence = (
            "L4 executed extended data checks; "
            f"loaded {loaded_count}/{len(selected)} files, rows_scanned={rows_scanned}, "
            f"missing_rate={missing_rate:.4f}, numeric_columns_total={numeric_cols_total}, domain_relevance={relevance_score:.2f}, "
            f"truncated_files={truncated_count}. Report: {artifact_path}"
        )
        return (
            status,
            score,
            {
                "data_root": str(data_root),
                "indexed_files": len(data_files),
                "selected_files": [str(p) for p in selected],
                "loaded_files": loaded_count,
                "rows_scanned": rows_scanned,
                "missing_rate": missing_rate,
                "numeric_columns_total": numeric_cols_total,
                "truncated_file_count": truncated_count,
                "operations_requested": op_names,
                "domain_relevance_score": relevance_score,
                "domain_relevance_evidence": relevance_evidence,
                "score_breakdown": score_components,
                "interpretability_summary": self._build_interpretability_summary(
                    level="L4",
                    loaded_count=loaded_count,
                    selected_count=len(selected),
                    rows_scanned=rows_scanned,
                    missing_rate=missing_rate,
                    numeric_cols_total=numeric_cols_total,
                    relevance_score=relevance_score,
                ),
                "biomni_call_trace": {
                    "registration_attempted": bool(self._registration_state),
                    "biomni_runtime_called": True,
                    "cache_hit": bool(probe.get("cache_hit", False)),
                    "biomni_probe_duration_ms": probe.get("duration_ms", 0),
                    "custom_tool_count": probe.get("custom_tool_count", 0),
                    "related_tool_hits": probe.get("related_tool_hits", 0),
                    "mcp_config_used": probe.get("mcp_config_used", str(self.mcp_config_path)),
                    "loaded_server_count": probe.get("loaded_server_count", -1),
                    "minimal_config_selected_count": probe.get("minimal_config_selected_count", 0),
                    "full_config_server_count": probe.get("full_config_server_count", 0),
                    "note": "L4 first performs a minimal Biomni runtime probe, then runs extended data checks.",
                },
                "downloaded_example_files": [str(p) for p in downloaded_examples],
                "l4_report_file": str(artifact_path),
                "biomni_execution": biomni_exec,
            },
            "" if biomni_exec.get("ok", False) else "biomni_execution_failed",
            evidence,
        )

    def validate_hypothesis(self, hypothesis: Hypothesis | dict[str, Any], level: str = "L1") -> dict[str, Any]:
        h = self._coerce_hypothesis(hypothesis)
        level = level.upper().strip()
        if level not in self.LEVEL_CONFIDENCE:
            level = "L1"

        if level == "L1":
            status, score, metrics, failure_reason, evidence = self._validate_l1(h)
        elif level == "L2":
            status, score, metrics, failure_reason, evidence = self._validate_l2(h)
        elif level == "L3":
            status, score, metrics, failure_reason, evidence = self._validate_l3(h)
        elif level == "L4":
            status, score, metrics, failure_reason, evidence = self._validate_l4(h)
        else:
            status, score, metrics, failure_reason, evidence = (
                "inconclusive",
                0.4,
                {"implemented": False},
                f"{level}_not_implemented_in_v1",
                f"{level} validation is scaffolded in first version.",
            )

        report = ValidationReport.build(
            hypothesis_id=h.get("hypothesis_id", ""),
            domain=h.get("domain", ""),
            level=level,
            status=status,
            score=score,
            confidence=self.LEVEL_CONFIDENCE[level],
            key_metrics=metrics,
            failure_reason=failure_reason,
            evidence=evidence,
        )
        self.memory.save_validation_report(report)
        # Save an insight so T1 can leverage execution outcomes as experience.
        try:
            title = f"{h.get('domain', 'general')} {level} {status}"
            recommendation = report.evidence
            biomni_exec = report.key_metrics.get("biomni_execution", {})
            if isinstance(biomni_exec, dict) and biomni_exec.get("recommendation"):
                recommendation = str(biomni_exec.get("recommendation"))
            ins = Insight.build(
                title=title,
                hypothesis=str(h.get("hypothesis", ""))[:1000],
                recommendation=recommendation[:2000],
                confidence=float(report.confidence),
                evidence_run_ids=[report.validation_id],
                tags=[str(h.get("domain", "general")), f"validation-{level.lower()}", status],
            )
            self.memory.save_insight(ins)
        except Exception:
            pass
        self.memory.update_summary_statistics()
        report_dict = report.to_dict()
        report_dict["runtime_report_file"] = self._persist_runtime_artifact(report_dict, h)
        return report_dict

    def validate_top_hypotheses(
        self,
        ranked_hypotheses: list[dict[str, Any]],
        top_m: int = 3,
        level: str = "L1",
    ) -> list[dict[str, Any]]:
        reports: list[dict[str, Any]] = []
        selected = ranked_hypotheses[: max(0, top_m)]
        total = len(selected)
        for idx, h in enumerate(selected, start=1):
            hid = h.get("hypothesis_id", f"idx_{idx}")
            print(f"[STAGE:VALIDATION] running {idx}/{total} level={level.upper()} hypothesis_id={hid}", flush=True)
            report = self.validate_hypothesis(h, level=level)
            print(
                f"[STAGE:VALIDATION] finished {idx}/{total} hypothesis_id={hid} status={report.get('status', 'unknown')} score={report.get('score', 0)}",
                flush=True,
            )
            reports.append(report)
        return reports