Blame for frontierlab/evaluation.py

"""Source-bound method comparison; researcher assertions are not execution evidence."""
from __future__ import annotations

from copy import deepcopy
from datetime import datetime, timezone
import hashlib
import json

from .models import Run
from .interventions import record_human_event
from .review import approved_claims, review_packet, source_fingerprint

CHECKS = ("data", "code", "environment", "parameters", "protocol", "metrics")
CHECK_STATES = ("unknown", "documented", "missing", "not_applicable")
CRITERIA = ("expected_value", "resources", "validation", "limitations", "novelty")
SCHEMA = "frontierlab.method-evaluation.v1"


def _digest(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False,
                                     separators=(",", ":"), allow_nan=False).encode()).hexdigest()


def allowed_sources(run: Run) -> list[str]:
    if not run.hypotheses:
        raise ValueError("Generate hypotheses before evaluating methods")
    if run.mode == "demo":
        return sorted(p.paper_id for p in run.papers if p.synthetic)
    if run.mode != "live":
        raise ValueError("Unsupported research mode")
    if not run.review or run.review.get("dataset_id") != review_packet(run)["dataset_id"]:
        raise PermissionError("Current source-bound human review required")
    ids = sorted({c.paper_id for c in approved_claims(run)})
    if run.review.get("status") != "READY_FOR_HUMAN_GATED_SYNTHESIS" or len(ids) < 2:
        raise PermissionError("Two human-supported sources and complete review required")
    if any(p.synthetic for p in run.papers):
        raise PermissionError("Synthetic publications cannot enter a live method evaluation")
    return ids


def context_id(run: Run) -> str:
    allowed_sources(run)
    return _digest({
        "question": run.question, "mode": run.mode, "model": run.model, "backend": run.backend,
        "sources": [source_fingerprint(p) for p in run.papers],
        "claims": [c.__dict__ for c in run.audit.verified_claims],
        "conflicts": run.audit.conflicts, "hypotheses": run.hypotheses,
        "human_review": run.review,
    })


def _text(value, label, minimum=1):
    if not isinstance(value, str) or not minimum <= len(value.strip()) <= 3000:
        raise ValueError(f"{label} requires {minimum}..3000 characters")
    return value.strip()


def blank_candidate(candidate_id: str, name: str = "") -> dict:
    return {
        "candidate_id": candidate_id, "name": name, "evidence_ids": [],
        "scope": "",
        "checks": {key: {"status": "unknown", "detail": "", "reference": ""}
                   for key in CHECKS},
        "comparison": {key: "" for key in CRITERIA},
    }


def _candidate(raw, known):
    if not isinstance(raw, dict):
        raise ValueError("Candidate must be an object")
    cid = _text(raw.get("candidate_id"), "Candidate ID")
    name = _text(raw.get("name"), "Method name")
    ids = raw.get("evidence_ids")
    if (not isinstance(ids, list) or not ids or
            any(not isinstance(x, str) or x not in known for x in ids)):
        raise ValueError("Method references must use reviewed source IDs")
    scope = _text(raw.get("scope"), "Applicability and scope")
    checks, gaps = {}, []
    supplied = raw.get("checks", {})
    if not isinstance(supplied, dict):
        raise ValueError("Reproducibility checks must be an object")
    for key in CHECKS:
        item = supplied.get(key, {})
        if not isinstance(item, dict) or item.get("status", "unknown") not in CHECK_STATES:
            raise ValueError("Invalid reproducibility check status")
        status = item.get("status", "unknown")
        detail = item.get("detail", "")
        reference = item.get("reference", "")
        if not isinstance(detail, str) or not isinstance(reference, str):
            raise ValueError("Check details and references must be text")
        if status != "unknown":
            detail = _text(detail, key + " explanation")
        if status == "documented":
            reference = _text(reference, key + " location/reference")
        checks[key] = {"status": status, "detail": detail.strip(), "reference": reference.strip()}
        if status in ("unknown", "missing"):
            gaps.append({"item": key, "status": status,
                         "reason": detail.strip() or "Not assessed; inspect original methods."})
    comparison = raw.get("comparison", {})
    if not isinstance(comparison, dict):
        raise ValueError("Comparison criteria must be an object")
    comparison = {key: _text(comparison.get(key), key) for key in CRITERIA}
    return {
        "candidate_id": cid, "name": name, "evidence_ids": sorted(set(ids)),
        "scope": scope, "checks": checks, "comparison": comparison,
        "reproducibility_status": "INCOMPLETE_DOCUMENTATION" if gaps else "DOCUMENTED_NOT_EXECUTED",
        "open_checks": gaps,
    }


def build_evaluation(run: Run, candidates: list[dict], *, selected_id: str | None,
                     rationale: str) -> dict:
    known = allowed_sources(run)
    if not isinstance(candidates, list) or not 0 <= len(candidates) <= 4:
        raise ValueError("Provide zero to four candidate methods")
    normalized = [_candidate(c, known) for c in candidates]
    ids = [c["candidate_id"] for c in normalized]
    if len(ids) != len(set(ids)):
        raise ValueError("Duplicate candidate IDs")
    rationale = _text(rationale, "Selection or deferral rationale", minimum=8)
    if selected_id is not None and (not isinstance(selected_id, str) or selected_id not in ids):
        raise ValueError("Select a known method or defer")
    if selected_id is not None and len(normalized) < 2:
        raise ValueError("At least two candidates required for a comparative selection; otherwise defer")
    status = "READY_FOR_DRAFT_PLANNING" if selected_id is not None else "DEFERRED_INSUFFICIENT_COMPARISON"
    result = {
        "schema": SCHEMA, "context_id": context_id(run), "status": status,
        "origin": "SYNTHETIC_FIXTURE" if run.mode == "demo" else "RESEARCHER_ASSESSED",
        "candidates": normalized, "selected_candidate_id": selected_id, "rationale": rationale,
        "uncertainty": {
            "source_count": len(known),
            "machine_flagged_tensions": deepcopy(run.audit.conflicts),
            "source_scope": "Abstract-level claim review; method records are researcher assertions.",
            "method_references_independently_checked": False,
            "confidence_calibrated": False,
            "scientific_execution_verified": False,
            "note": "Selection is a reasoned draft decision, not a probability of success or proven superiority.",
        },
    }
    result["evaluation_id"] = _digest(result)
    return result


def save_evaluation(run: Run, candidates: list[dict], *, selected_id: str | None,
                    rationale: str) -> dict:
    result = build_evaluation(run, candidates, selected_id=selected_id, rationale=rationale)
    run.evaluation = result
    run.plan = None
    run.log.append({"time_utc": datetime.now(timezone.utc).isoformat(),
                    "role": "Method Evaluation", "status": result["status"],
                    "note": "Researcher-saved method comparison; evaluation_id=" + result["evaluation_id"]})
    record_human_event(run, "METHOD_EVALUATION_SAVED", result["evaluation_id"], note=result["status"])
    return result


def require_evaluation(run: Run) -> dict:
    saved = run.evaluation
    if not isinstance(saved, dict) or saved.get("schema") != SCHEMA:
        raise PermissionError("Complete method evaluation before planning")
    rebuilt = build_evaluation(run, saved.get("candidates"),
                               selected_id=saved.get("selected_candidate_id"),
                               rationale=saved.get("rationale"))
    if rebuilt != saved:
        raise PermissionError("Method evaluation is stale or changed; review it again")
    if saved["status"] != "READY_FOR_DRAFT_PLANNING":
        raise PermissionError("Method comparison was deferred; planning remains blocked")
    return saved


def selected_method(evaluation: dict) -> dict:
    return next(c for c in evaluation["candidates"]
                if c["candidate_id"] == evaluation["selected_candidate_id"])


def suggest_methods(run: Run, *, model_client) -> list[dict]:
    """One optional model call; output remains an editable, unreviewed proposal."""
    known = allowed_sources(run)
    if run.mode != "live" or model_client is None:
        raise PermissionError("Method suggestions require a live, connected model")
    if run.backend not in ("none", "unspecified") and getattr(model_client, "backend", "unspecified") != run.backend:
        raise PermissionError("Use this run's original model backend for method suggestions")
    response = model_client.json("Method Evaluation Agent", (
        'Propose up to two alternative research methods relevant to the hypotheses. '
        'Return {"candidates":[{"name":string,"evidence_ids":[provided IDs],"scope":string,'
        '"comparison":{"expected_value":string,"resources":string,"validation":string,'
        '"limitations":string,"novelty":string}}]}. '
        'Use only provided abstracts and reviewed claims. Say UNKNOWN when details, '
        'novelty or resources cannot be inferred. Do not claim code was executed or '
        'that structure confidence establishes binding, pH selectivity or agonism. '
        'Do not treat complementary pipeline tools as competing methods. '
        'Return an empty candidates list if no meaningful alternatives are supported.'
    ), {"question": run.question, "hypotheses": run.hypotheses,
        "papers": [p.__dict__ for p in run.papers if p.paper_id in known],
        "reviewed_claims": [c.__dict__ for c in approved_claims(run)]})
    raw = response.get("candidates")
    if not isinstance(raw, list) or len(raw) > 2:
        raise ValueError("Expected zero to two method proposals")
    out = []
    for i, item in enumerate(raw, 1):
        if not isinstance(item, dict):
            raise ValueError("Invalid method proposal")
        candidate = blank_candidate(f"M{i}", item.get("name", ""))
        candidate.update({key: item.get(key) for key in ("evidence_ids", "scope", "comparison")})
        checked = _candidate(candidate, known)
        out.append({key: checked[key] for key in candidate})
    return out


def demo_candidates() -> list[dict]:
    """Invented strategies; no real experiment or method comparison was run."""
    out = []
    for cid, name, scope, benefit in [
        ("M1", "[FICTIONAL] Matched-condition replication",
         "Compare the same invented binder under matched buffer and assay conditions.",
         "Could separate a buffer confound from an apparent pH effect."),
        ("M2", "[FICTIONAL] Cross-assay confirmation",
         "Compare the invented observation across independent assay formats.",
         "Could expose assay-format dependence before interpreting the pH response."),
    ]:
        c = blank_candidate(cid, name)
        c.update(evidence_ids=["SYNTHETIC:A", "SYNTHETIC:B", "SYNTHETIC:C"], scope=scope)
        c["comparison"] = {
            "expected_value": benefit,
            "resources": "Unknown; obtain protocol, materials and access estimates.",
            "validation": "Require independent controls and repeats; no experiment has run.",
            "limitations": "Entirely fictional abstracts; actual methods and effect sizes unavailable.",
            "novelty": "Not assessed; these are illustrative validation strategies.",
        }
        out.append(c)
    return out