Hypothesis Generation

scripts/validate_hypothesis_schema.py

#!/usr/bin/env python3
"""Validate a structured hypothesis record without judging scientific merit."""

from __future__ import annotations

import argparse
from collections import Counter
from pathlib import PurePosixPath
from typing import Any

from _common import (
    ValidationError,
    error_exit,
    issue,
    read_json,
    require_bool,
    require_enum,
    require_exact_keys,
    require_identifier,
    require_identifier_list,
    require_iso_date,
    require_list,
    require_object,
    require_text,
    require_text_list,
    require_unique,
    write_json_report,
)

QUESTION_TYPES = {"descriptive", "associational", "predictive", "causal", "mechanistic"}
HYPOTHESIS_STATUS = {"candidate"}
MEASUREMENT_ROLES = {
    "intervention",
    "exposure",
    "outcome",
    "mediator",
    "confounder",
    "selection",
    "negative_control",
    "positive_control",
    "effect_modifier",
    "other",
}
RISK_TYPES = {
    "confounding",
    "selection_bias",
    "collider_bias",
    "reverse_causation",
    "measurement_bias",
    "other",
}
CONTROL_TYPES = {
    "negative_exposure",
    "negative_outcome",
    "procedural_negative",
    "positive_control",
    "vehicle_or_sham",
    "other",
}
PLAN_MODES = {"confirmatory", "exploratory"}
GATE_STATES = {"not_applicable", "undetermined", "requires_review", "approved", "blocked"}
FEASIBILITY_STATES = {
    "undetermined",
    "feasible_for_planning",
    "requires_pilot",
    "infeasible",
    "blocked",
}


def _parse_observation(raw: Any) -> dict[str, Any]:
    value = require_object(raw, "record.observation")
    require_exact_keys(
        value,
        required={"statement", "provenance", "source_ids", "uncertainties"},
        context="record.observation",
    )
    return {
        "statement": require_text(
            value["statement"], "record.observation.statement", minimum=10
        ),
        "provenance": require_text(
            value["provenance"], "record.observation.provenance", minimum=10
        ),
        "source_ids": require_identifier_list(
            value["source_ids"], "record.observation.source_ids", minimum=1
        ),
        "uncertainties": require_text_list(
            value["uncertainties"],
            "record.observation.uncertainties",
            minimum=1,
        ),
    }


def _parse_question(raw: Any) -> dict[str, str]:
    value = require_object(raw, "record.research_question")
    fields = {
        "statement",
        "framework",
        "question_type",
        "population_or_system",
        "intervention_or_exposure",
        "comparator",
        "outcome",
        "timeframe",
    }
    require_exact_keys(value, required=fields, context="record.research_question")
    parsed = {
        field: require_text(
            value[field],
            f"record.research_question.{field}",
            minimum=2 if field == "framework" else 5,
        )
        for field in fields
        if field != "question_type"
    }
    parsed["question_type"] = require_enum(
        value["question_type"],
        QUESTION_TYPES,
        "record.research_question.question_type",
    )
    return parsed


def _parse_hypotheses(raw: Any) -> list[dict[str, Any]]:
    entries = require_list(raw, "record.hypotheses", minimum=1, maximum=50)
    parsed: list[dict[str, Any]] = []
    identifiers: list[str] = []
    fields = {
        "hypothesis_id",
        "statement",
        "mechanism",
        "status",
        "source_ids",
        "assumptions",
        "boundary_conditions",
        "uncertainties",
        "prediction_ids",
        "rival_hypothesis_ids",
    }
    for index, raw_entry in enumerate(entries):
        context = f"record.hypotheses[{index}]"
        entry = require_object(raw_entry, context)
        require_exact_keys(entry, required=fields, context=context)
        identifier = require_identifier(entry["hypothesis_id"], f"{context}.hypothesis_id")
        identifiers.append(identifier)
        parsed.append(
            {
                "hypothesis_id": identifier,
                "statement": require_text(
                    entry["statement"], f"{context}.statement", minimum=10
                ),
                "mechanism": require_text(
                    entry["mechanism"], f"{context}.mechanism", minimum=10
                ),
                "status": require_enum(
                    entry["status"], HYPOTHESIS_STATUS, f"{context}.status"
                ),
                "source_ids": require_identifier_list(
                    entry["source_ids"], f"{context}.source_ids", minimum=1
                ),
                "assumptions": require_text_list(
                    entry["assumptions"], f"{context}.assumptions", minimum=1
                ),
                "boundary_conditions": require_text_list(
                    entry["boundary_conditions"],
                    f"{context}.boundary_conditions",
                    minimum=1,
                ),
                "uncertainties": require_text_list(
                    entry["uncertainties"], f"{context}.uncertainties", minimum=1
                ),
                "prediction_ids": require_identifier_list(
                    entry["prediction_ids"], f"{context}.prediction_ids", minimum=1
                ),
                "rival_hypothesis_ids": require_identifier_list(
                    entry["rival_hypothesis_ids"],
                    f"{context}.rival_hypothesis_ids",
                ),
            }
        )
    require_unique(identifiers, "record.hypotheses")
    return parsed


def _parse_estimands(raw: Any) -> list[dict[str, Any]]:
    entries = require_list(raw, "record.causal_estimands", maximum=50)
    parsed: list[dict[str, Any]] = []
    identifiers: list[str] = []
    fields = {
        "estimand_id",
        "linked_hypothesis_ids",
        "population",
        "intervention_or_exposure",
        "comparator",
        "outcome",
        "time_horizon",
        "population_summary",
        "intercurrent_event_strategy",
        "identification_assumptions",
    }
    for index, raw_entry in enumerate(entries):
        context = f"record.causal_estimands[{index}]"
        entry = require_object(raw_entry, context)
        require_exact_keys(entry, required=fields, context=context)
        identifier = require_identifier(entry["estimand_id"], f"{context}.estimand_id")
        identifiers.append(identifier)
        parsed.append(
            {
                "estimand_id": identifier,
                "linked_hypothesis_ids": require_identifier_list(
                    entry["linked_hypothesis_ids"],
                    f"{context}.linked_hypothesis_ids",
                    minimum=1,
                ),
                "population": require_text(entry["population"], f"{context}.population"),
                "intervention_or_exposure": require_text(
                    entry["intervention_or_exposure"],
                    f"{context}.intervention_or_exposure",
                ),
                "comparator": require_text(
                    entry["comparator"], f"{context}.comparator"
                ),
                "outcome": require_text(entry["outcome"], f"{context}.outcome"),
                "time_horizon": require_text(
                    entry["time_horizon"], f"{context}.time_horizon"
                ),
                "population_summary": require_text(
                    entry["population_summary"], f"{context}.population_summary"
                ),
                "intercurrent_event_strategy": require_text(
                    entry["intercurrent_event_strategy"],
                    f"{context}.intercurrent_event_strategy",
                ),
                "identification_assumptions": require_text_list(
                    entry["identification_assumptions"],
                    f"{context}.identification_assumptions",
                    minimum=1,
                ),
            }
        )
    require_unique(identifiers, "record.causal_estimands")
    return parsed


def _parse_predictions(raw: Any) -> list[dict[str, Any]]:
    entries = require_list(raw, "record.predictions", minimum=1, maximum=200)
    parsed: list[dict[str, Any]] = []
    identifiers: list[str] = []
    fields = {
        "prediction_id",
        "hypothesis_id",
        "statement",
        "observable",
        "conditions",
        "expected_pattern",
        "falsifier",
        "rival_hypothesis_ids",
        "measurement_ids",
        "analysis_ids",
    }
    for index, raw_entry in enumerate(entries):
        context = f"record.predictions[{index}]"
        entry = require_object(raw_entry, context)
        require_exact_keys(entry, required=fields, context=context)
        identifier = require_identifier(entry["prediction_id"], f"{context}.prediction_id")
        identifiers.append(identifier)
        parsed.append(
            {
                "prediction_id": identifier,
                "hypothesis_id": require_identifier(
                    entry["hypothesis_id"], f"{context}.hypothesis_id"
                ),
                "statement": require_text(
                    entry["statement"], f"{context}.statement", minimum=10
                ),
                "observable": require_text(
                    entry["observable"], f"{context}.observable", minimum=5
                ),
                "conditions": require_text(
                    entry["conditions"], f"{context}.conditions", minimum=5
                ),
                "expected_pattern": require_text(
                    entry["expected_pattern"], f"{context}.expected_pattern", minimum=5
                ),
                "falsifier": require_text(
                    entry["falsifier"], f"{context}.falsifier", minimum=10
                ),
                "rival_hypothesis_ids": require_identifier_list(
                    entry["rival_hypothesis_ids"],
                    f"{context}.rival_hypothesis_ids",
                    minimum=1,
                ),
                "measurement_ids": require_identifier_list(
                    entry["measurement_ids"],
                    f"{context}.measurement_ids",
                    minimum=1,
                ),
                "analysis_ids": require_identifier_list(
                    entry["analysis_ids"], f"{context}.analysis_ids", minimum=1
                ),
            }
        )
    require_unique(identifiers, "record.predictions")
    return parsed


def _parse_alternatives(raw: Any) -> list[dict[str, Any]]:
    entries = require_list(raw, "record.alternative_explanations", minimum=1, maximum=100)
    parsed: list[dict[str, Any]] = []
    identifiers: list[str] = []
    fields = {
        "alternative_id",
        "statement",
        "linked_hypothesis_ids",
        "risk_types",
        "discriminating_prediction_ids",
    }
    for index, raw_entry in enumerate(entries):
        context = f"record.alternative_explanations[{index}]"
        entry = require_object(raw_entry, context)
        require_exact_keys(entry, required=fields, context=context)
        identifier = require_identifier(entry["alternative_id"], f"{context}.alternative_id")
        identifiers.append(identifier)
        risk_types = require_text_list(
            entry["risk_types"], f"{context}.risk_types", minimum=1, maximum=10
        )
        for risk_index, risk_type in enumerate(risk_types):
            require_enum(risk_type, RISK_TYPES, f"{context}.risk_types[{risk_index}]")
        require_unique(risk_types, f"{context}.risk_types")
        parsed.append(
            {
                "alternative_id": identifier,
                "statement": require_text(
                    entry["statement"], f"{context}.statement", minimum=10
                ),
                "linked_hypothesis_ids": require_identifier_list(
                    entry["linked_hypothesis_ids"],
                    f"{context}.linked_hypothesis_ids",
                    minimum=1,
                ),
                "risk_types": risk_types,
                "discriminating_prediction_ids": require_identifier_list(
                    entry["discriminating_prediction_ids"],
                    f"{context}.discriminating_prediction_ids",
                    minimum=1,
                ),
            }
        )
    require_unique(identifiers, "record.alternative_explanations")
    return parsed


def _parse_nulls(raw: Any) -> list[dict[str, Any]]:
    entries = require_list(raw, "record.null_hypotheses", minimum=1, maximum=100)
    parsed: list[dict[str, Any]] = []
    identifiers: list[str] = []
    fields = {
        "null_id",
        "statement",
        "linked_prediction_ids",
        "rejection_or_compatibility_rule",
    }
    for index, raw_entry in enumerate(entries):
        context = f"record.null_hypotheses[{index}]"
        entry = require_object(raw_entry, context)
        require_exact_keys(entry, required=fields, context=context)
        identifier = require_identifier(entry["null_id"], f"{context}.null_id")
        identifiers.append(identifier)
        parsed.append(
            {
                "null_id": identifier,
                "statement": require_text(
                    entry["statement"], f"{context}.statement", minimum=10
                ),
                "linked_prediction_ids": require_identifier_list(
                    entry["linked_prediction_ids"],
                    f"{context}.linked_prediction_ids",
                    minimum=1,
                ),
                "rejection_or_compatibility_rule": require_text(
                    entry["rejection_or_compatibility_rule"],
                    f"{context}.rejection_or_compatibility_rule",
                    minimum=10,
                ),
            }
        )
    require_unique(identifiers, "record.null_hypotheses")
    return parsed


def _parse_controls(raw: Any) -> list[dict[str, Any]]:
    entries = require_list(raw, "record.negative_controls", minimum=1, maximum=100)
    parsed: list[dict[str, Any]] = []
    identifiers: list[str] = []
    fields = {
        "control_id",
        "control_type",
        "rationale",
        "expected_result",
        "failure_implication",
        "linked_prediction_ids",
    }
    for index, raw_entry in enumerate(entries):
        context = f"record.negative_controls[{index}]"
        entry = require_object(raw_entry, context)
        require_exact_keys(entry, required=fields, context=context)
        identifier = require_identifier(entry["control_id"], f"{context}.control_id")
        identifiers.append(identifier)
        parsed.append(
            {
                "control_id": identifier,
                "control_type": require_enum(
                    entry["control_type"], CONTROL_TYPES, f"{context}.control_type"
                ),
                "rationale": require_text(
                    entry["rationale"], f"{context}.rationale", minimum=10
                ),
                "expected_result": require_text(
                    entry["expected_result"], f"{context}.expected_result", minimum=5
                ),
                "failure_implication": require_text(
                    entry["failure_implication"],
                    f"{context}.failure_implication",
                    minimum=10,
                ),
                "linked_prediction_ids": require_identifier_list(
                    entry["linked_prediction_ids"],
                    f"{context}.linked_prediction_ids",
                    minimum=1,
                ),
            }
        )
    require_unique(identifiers, "record.negative_controls")
    return parsed


def _parse_operationalizations(raw: Any) -> list[dict[str, Any]]:
    entries = require_list(raw, "record.operationalizations", minimum=1, maximum=200)
    parsed: list[dict[str, Any]] = []
    identifiers: list[str] = []
    fields = {
        "measurement_id",
        "construct",
        "variable",
        "role",
        "operational_definition",
        "instrument_or_method",
        "unit",
        "timing",
        "population_or_system",
        "validity_evidence_source_ids",
        "reliability_plan",
        "missingness_plan",
        "blinding_or_masking",
        "threshold_rationale",
    }
    for index, raw_entry in enumerate(entries):
        context = f"record.operationalizations[{index}]"
        entry = require_object(raw_entry, context)
        require_exact_keys(entry, required=fields, context=context)
        identifier = require_identifier(entry["measurement_id"], f"{context}.measurement_id")
        identifiers.append(identifier)
        text_fields = fields - {
            "measurement_id",
            "role",
            "validity_evidence_source_ids",
        }
        parsed_entry: dict[str, Any] = {
            "measurement_id": identifier,
            "role": require_enum(entry["role"], MEASUREMENT_ROLES, f"{context}.role"),
            "validity_evidence_source_ids": require_identifier_list(
                entry["validity_evidence_source_ids"],
                f"{context}.validity_evidence_source_ids",
                minimum=1,
            ),
        }
        for field in text_fields:
            parsed_entry[field] = require_text(
                entry[field],
                f"{context}.{field}",
                minimum=2 if field in {"unit", "variable"} else 5,
            )
        parsed.append(parsed_entry)
    require_unique(identifiers, "record.operationalizations")
    return parsed


def _parse_analysis_plan(raw: Any) -> dict[str, Any]:
    plan = require_object(raw, "record.analysis_plan")
    require_exact_keys(
        plan,
        required={"analyses", "harking_control", "deviation_reporting"},
        context="record.analysis_plan",
    )
    entries = require_list(
        plan["analyses"], "record.analysis_plan.analyses", minimum=1, maximum=200
    )
    parsed: list[dict[str, Any]] = []
    identifiers: list[str] = []
    fields = {
        "analysis_id",
        "prediction_ids",
        "estimand_ids",
        "analysis_population",
        "method",
        "effect_or_summary_measure",
        "uncertainty_method",
        "missing_data_plan",
        "multiplicity_plan",
        "sensitivity_analyses",
        "decision_rule",
        "exploratory_or_confirmatory",
    }
    for index, raw_entry in enumerate(entries):
        context = f"record.analysis_plan.analyses[{index}]"
        entry = require_object(raw_entry, context)
        require_exact_keys(entry, required=fields, context=context)
        identifier = require_identifier(entry["analysis_id"], f"{context}.analysis_id")
        identifiers.append(identifier)
        parsed.append(
            {
                "analysis_id": identifier,
                "prediction_ids": require_identifier_list(
                    entry["prediction_ids"], f"{context}.prediction_ids", minimum=1
                ),
                "estimand_ids": require_identifier_list(
                    entry["estimand_ids"], f"{context}.estimand_ids"
                ),
                "analysis_population": require_text(
                    entry["analysis_population"],
                    f"{context}.analysis_population",
                    minimum=5,
                ),
                "method": require_text(entry["method"], f"{context}.method", minimum=5),
                "effect_or_summary_measure": require_text(
                    entry["effect_or_summary_measure"],
                    f"{context}.effect_or_summary_measure",
                    minimum=3,
                ),
                "uncertainty_method": require_text(
                    entry["uncertainty_method"],
                    f"{context}.uncertainty_method",
                    minimum=5,
                ),
                "missing_data_plan": require_text(
                    entry["missing_data_plan"],
                    f"{context}.missing_data_plan",
                    minimum=5,
                ),
                "multiplicity_plan": require_text(
                    entry["multiplicity_plan"],
                    f"{context}.multiplicity_plan",
                    minimum=5,
                ),
                "sensitivity_analyses": require_text_list(
                    entry["sensitivity_analyses"],
                    f"{context}.sensitivity_analyses",
                    minimum=1,
                ),
                "decision_rule": require_text(
                    entry["decision_rule"], f"{context}.decision_rule", minimum=10
                ),
                "exploratory_or_confirmatory": require_enum(
                    entry["exploratory_or_confirmatory"],
                    PLAN_MODES,
                    f"{context}.exploratory_or_confirmatory",
                ),
            }
        )
    require_unique(identifiers, "record.analysis_plan.analyses")
    return {
        "analyses": parsed,
        "harking_control": require_text(
            plan["harking_control"],
            "record.analysis_plan.harking_control",
            minimum=10,
        ),
        "deviation_reporting": require_text(
            plan["deviation_reporting"],
            "record.analysis_plan.deviation_reporting",
            minimum=10,
        ),
    }


def _parse_evidence(raw: Any) -> dict[str, Any]:
    value = require_object(raw, "record.evidence")
    require_exact_keys(
        value,
        required={
            "ledger_path",
            "source_ids",
            "search_boundary_id",
            "evidence_limitations",
        },
        context="record.evidence",
    )
    ledger_path = require_text(
        value["ledger_path"], "record.evidence.ledger_path", maximum=500
    )
    pure_path = PurePosixPath(ledger_path)
    if pure_path.is_absolute() or ".." in pure_path.parts or "://" in ledger_path:
        raise ValidationError("record.evidence.ledger_path must be a safe relative path")
    if pure_path.suffix.lower() != ".csv":
        raise ValidationError("record.evidence.ledger_path must end in .csv")
    return {
        "ledger_path": ledger_path,
        "source_ids": require_identifier_list(
            value["source_ids"], "record.evidence.source_ids", minimum=1
        ),
        "search_boundary_id": require_identifier(
            value["search_boundary_id"], "record.evidence.search_boundary_id"
        ),
        "evidence_limitations": require_text_list(
            value["evidence_limitations"],
            "record.evidence.evidence_limitations",
            minimum=1,
        ),
    }


def _parse_risk_register(raw: Any) -> dict[str, list[str]]:
    value = require_object(raw, "record.risk_register")
    fields = {
        "confounding",
        "selection_bias",
        "collider_bias",
        "reverse_causation",
        "measurement_bias",
        "other",
    }
    require_exact_keys(value, required=fields, context="record.risk_register")
    return {
        field: require_text_list(
            value[field], f"record.risk_register.{field}", maximum=50
        )
        for field in fields
    }


def _parse_ethics(raw: Any) -> dict[str, Any]:
    value = require_object(raw, "record.ethics_and_feasibility")
    gate_fields = {
        "human_subjects_gate",
        "animal_research_gate",
        "biosafety_gate",
        "dual_use_gate",
        "regulatory_gate",
        "data_governance_gate",
    }
    require_exact_keys(
        value,
        required=gate_fields
        | {"feasibility_status", "required_reviews", "unresolved_blocks"},
        context="record.ethics_and_feasibility",
    )
    parsed: dict[str, Any] = {
        field: require_enum(
            value[field], GATE_STATES, f"record.ethics_and_feasibility.{field}"
        )
        for field in gate_fields
    }
    parsed["feasibility_status"] = require_enum(
        value["feasibility_status"],
        FEASIBILITY_STATES,
        "record.ethics_and_feasibility.feasibility_status",
    )
    parsed["required_reviews"] = require_text_list(
        value["required_reviews"],
        "record.ethics_and_feasibility.required_reviews",
        maximum=50,
    )
    parsed["unresolved_blocks"] = require_text_list(
        value["unresolved_blocks"],
        "record.ethics_and_feasibility.unresolved_blocks",
        maximum=50,
    )
    return parsed


def _parse_ai_use(raw: Any) -> dict[str, Any]:
    value = require_object(raw, "record.ai_use")
    fields = {
        "used",
        "sensitive_or_unpublished_data_sent_externally",
        "local_policy_checked",
        "citation_verification_required",
        "human_accountable",
        "diversity_mitigation",
    }
    require_exact_keys(value, required=fields, context="record.ai_use")
    return {
        "used": require_bool(value["used"], "record.ai_use.used"),
        "sensitive_or_unpublished_data_sent_externally": require_bool(
            value["sensitive_or_unpublished_data_sent_externally"],
            "record.ai_use.sensitive_or_unpublished_data_sent_externally",
        ),
        "local_policy_checked": require_bool(
            value["local_policy_checked"], "record.ai_use.local_policy_checked"
        ),
        "citation_verification_required": require_bool(
            value["citation_verification_required"],
            "record.ai_use.citation_verification_required",
        ),
        "human_accountable": require_bool(
            value["human_accountable"], "record.ai_use.human_accountable"
        ),
        "diversity_mitigation": require_text(
            value["diversity_mitigation"],
            "record.ai_use.diversity_mitigation",
            minimum=10,
        ),
    }


def load_hypothesis_record(payload: Any) -> dict[str, Any]:
    """Parse the exact v2 record schema and normalize bounded values."""
    root = require_object(payload, "record")
    fields = {
        "schema_version",
        "project_id",
        "status",
        "updated_on",
        "human_owner",
        "observation",
        "research_question",
        "hypotheses",
        "causal_estimands",
        "predictions",
        "alternative_explanations",
        "null_hypotheses",
        "negative_controls",
        "operationalizations",
        "analysis_plan",
        "evidence",
        "risk_register",
        "ethics_and_feasibility",
        "ai_use",
    }
    require_exact_keys(root, required=fields, context="record")
    return {
        "schema_version": require_enum(
            root["schema_version"], {"2.0"}, "record.schema_version"
        ),
        "project_id": require_identifier(root["project_id"], "record.project_id"),
        "status": require_enum(
            root["status"], {"draft", "preregistered", "archived"}, "record.status"
        ),
        "updated_on": require_iso_date(root["updated_on"], "record.updated_on"),
        "human_owner": require_text(
            root["human_owner"], "record.human_owner", minimum=3
        ),
        "observation": _parse_observation(root["observation"]),
        "research_question": _parse_question(root["research_question"]),
        "hypotheses": _parse_hypotheses(root["hypotheses"]),
        "causal_estimands": _parse_estimands(root["causal_estimands"]),
        "predictions": _parse_predictions(root["predictions"]),
        "alternative_explanations": _parse_alternatives(
            root["alternative_explanations"]
        ),
        "null_hypotheses": _parse_nulls(root["null_hypotheses"]),
        "negative_controls": _parse_controls(root["negative_controls"]),
        "operationalizations": _parse_operationalizations(root["operationalizations"]),
        "analysis_plan": _parse_analysis_plan(root["analysis_plan"]),
        "evidence": _parse_evidence(root["evidence"]),
        "risk_register": _parse_risk_register(root["risk_register"]),
        "ethics_and_feasibility": _parse_ethics(root["ethics_and_feasibility"]),
        "ai_use": _parse_ai_use(root["ai_use"]),
    }


def _missing_references(
    values: list[str], valid_values: set[str], code: str, field: str
) -> list[dict[str, str]]:
    return [
        issue(code, f"{field}:{value}")
        for value in sorted(set(values) - valid_values)
    ]


def validate_record(record: dict[str, Any]) -> dict[str, Any]:
    """Check cross-links and safety declarations without scoring candidates."""
    errors: list[dict[str, str]] = []
    warnings: list[dict[str, str]] = []

    hypothesis_ids = {item["hypothesis_id"] for item in record["hypotheses"]}
    estimand_ids = {item["estimand_id"] for item in record["causal_estimands"]}
    prediction_ids = {item["prediction_id"] for item in record["predictions"]}
    measurement_ids = {
        item["measurement_id"] for item in record["operationalizations"]
    }
    analysis_ids = {
        item["analysis_id"] for item in record["analysis_plan"]["analyses"]
    }
    control_ids = {item["control_id"] for item in record["negative_controls"]}
    source_ids = set(record["evidence"]["source_ids"])

    used_sources = set(record["observation"]["source_ids"])
    for hypothesis in record["hypotheses"]:
        used_sources.update(hypothesis["source_ids"])
        errors.extend(
            _missing_references(
                hypothesis["prediction_ids"],
                prediction_ids,
                "UNKNOWN_PREDICTION_ID",
                hypothesis["hypothesis_id"],
            )
        )
        errors.extend(
            _missing_references(
                hypothesis["rival_hypothesis_ids"],
                hypothesis_ids,
                "UNKNOWN_RIVAL_HYPOTHESIS_ID",
                hypothesis["hypothesis_id"],
            )
        )
        if hypothesis["hypothesis_id"] in hypothesis["rival_hypothesis_ids"]:
            errors.append(
                issue("HYPOTHESIS_CANNOT_RIVAL_ITSELF", hypothesis["hypothesis_id"])
            )
        if len(hypothesis_ids) > 1 and not hypothesis["rival_hypothesis_ids"]:
            warnings.append(
                issue("RIVAL_HYPOTHESIS_LINK_MISSING", hypothesis["hypothesis_id"])
            )
    for operationalization in record["operationalizations"]:
        used_sources.update(operationalization["validity_evidence_source_ids"])
    errors.extend(
        _missing_references(
            sorted(used_sources),
            source_ids,
            "SOURCE_NOT_DECLARED_IN_EVIDENCE",
            "record.evidence.source_ids",
        )
    )

    if len(hypothesis_ids) == 1:
        warnings.append(issue("SINGLE_CANDIDATE_REQUIRES_RIVAL_REVIEW", "hypotheses"))

    for estimand in record["causal_estimands"]:
        errors.extend(
            _missing_references(
                estimand["linked_hypothesis_ids"],
                hypothesis_ids,
                "UNKNOWN_HYPOTHESIS_ID",
                estimand["estimand_id"],
            )
        )
    if (
        record["research_question"]["question_type"] == "causal"
        and not record["causal_estimands"]
    ):
        errors.append(issue("CAUSAL_QUESTION_REQUIRES_ESTIMAND", "causal_estimands"))

    predictions_by_hypothesis: Counter[str] = Counter()
    for prediction in record["predictions"]:
        prediction_id = prediction["prediction_id"]
        predictions_by_hypothesis[prediction["hypothesis_id"]] += 1
        errors.extend(
            _missing_references(
                [prediction["hypothesis_id"]],
                hypothesis_ids,
                "UNKNOWN_HYPOTHESIS_ID",
                prediction_id,
            )
        )
        errors.extend(
            _missing_references(
                prediction["rival_hypothesis_ids"],
                hypothesis_ids,
                "UNKNOWN_RIVAL_HYPOTHESIS_ID",
                prediction_id,
            )
        )
        errors.extend(
            _missing_references(
                prediction["measurement_ids"],
                measurement_ids,
                "UNKNOWN_MEASUREMENT_ID",
                prediction_id,
            )
        )
        errors.extend(
            _missing_references(
                prediction["analysis_ids"],
                analysis_ids,
                "UNKNOWN_ANALYSIS_ID",
                prediction_id,
            )
        )
        if prediction["hypothesis_id"] in prediction["rival_hypothesis_ids"]:
            errors.append(issue("PREDICTION_RIVAL_IS_FOCAL", prediction_id))
    for hypothesis_id in sorted(hypothesis_ids):
        if predictions_by_hypothesis[hypothesis_id] == 0:
            errors.append(issue("HYPOTHESIS_HAS_NO_PREDICTION", hypothesis_id))

    for alternative in record["alternative_explanations"]:
        errors.extend(
            _missing_references(
                alternative["linked_hypothesis_ids"],
                hypothesis_ids,
                "UNKNOWN_HYPOTHESIS_ID",
                alternative["alternative_id"],
            )
        )
        errors.extend(
            _missing_references(
                alternative["discriminating_prediction_ids"],
                prediction_ids,
                "UNKNOWN_PREDICTION_ID",
                alternative["alternative_id"],
            )
        )

    for null in record["null_hypotheses"]:
        errors.extend(
            _missing_references(
                null["linked_prediction_ids"],
                prediction_ids,
                "UNKNOWN_PREDICTION_ID",
                null["null_id"],
            )
        )

    for control in record["negative_controls"]:
        errors.extend(
            _missing_references(
                control["linked_prediction_ids"],
                prediction_ids,
                "UNKNOWN_PREDICTION_ID",
                control["control_id"],
            )
        )
    if not any(
        control["control_type"]
        in {"negative_exposure", "negative_outcome", "procedural_negative"}
        for control in record["negative_controls"]
    ):
        warnings.append(issue("NEGATIVE_CONTROL_TYPE_REVIEW_REQUIRED", "controls"))

    for analysis in record["analysis_plan"]["analyses"]:
        errors.extend(
            _missing_references(
                analysis["prediction_ids"],
                prediction_ids,
                "UNKNOWN_PREDICTION_ID",
                analysis["analysis_id"],
            )
        )
        errors.extend(
            _missing_references(
                analysis["estimand_ids"],
                estimand_ids,
                "UNKNOWN_ESTIMAND_ID",
                analysis["analysis_id"],
            )
        )

    for category, entries in record["risk_register"].items():
        if not entries:
            warnings.append(issue("RISK_CATEGORY_EMPTY_REQUIRES_RATIONALE", category))

    ethics = record["ethics_and_feasibility"]
    gate_values = {
        key: value for key, value in ethics.items() if key.endswith("_gate")
    }
    unresolved_gate_names = sorted(
        key
        for key, value in gate_values.items()
        if value in {"undetermined", "requires_review", "blocked"}
    )
    if unresolved_gate_names and not ethics["unresolved_blocks"]:
        errors.append(
            issue("UNRESOLVED_GATE_REQUIRES_BLOCK_RECORD", "ethics_and_feasibility")
        )
    if (
        any(value == "requires_review" for value in gate_values.values())
        and not ethics["required_reviews"]
    ):
        errors.append(
            issue("REQUIRED_REVIEW_LIST_MISSING", "ethics_and_feasibility")
        )
    if ethics["feasibility_status"] in {"undetermined", "infeasible", "blocked"}:
        unresolved_gate_names.append("feasibility_status")

    ai_use = record["ai_use"]
    if ai_use["sensitive_or_unpublished_data_sent_externally"]:
        errors.append(
            issue(
                "EXTERNAL_SENSITIVE_DATA_NOT_SUPPORTED",
                "ai_use.sensitive_or_unpublished_data_sent_externally",
            )
        )
    if not ai_use["local_policy_checked"]:
        errors.append(issue("LOCAL_AI_POLICY_NOT_CHECKED", "ai_use.local_policy_checked"))
    if not ai_use["citation_verification_required"]:
        errors.append(
            issue(
                "CITATION_VERIFICATION_MUST_BE_REQUIRED",
                "ai_use.citation_verification_required",
            )
        )
    if not ai_use["human_accountable"]:
        errors.append(
            issue("HUMAN_ACCOUNTABILITY_REQUIRED", "ai_use.human_accountable")
        )

    return {
        "schema_version": "2.0",
        "project_id": record["project_id"],
        "valid": not errors,
        "status": (
            "INVALID_RECORD"
            if errors
            else "VALID_BLOCKED_BY_GATES"
            if unresolved_gate_names
            else "VALID_FOR_HUMAN_REVIEW"
        ),
        "errors": errors,
        "warnings": warnings,
        "counts": {
            "hypotheses": len(hypothesis_ids),
            "causal_estimands": len(estimand_ids),
            "predictions": len(prediction_ids),
            "alternative_explanations": len(record["alternative_explanations"]),
            "null_hypotheses": len(record["null_hypotheses"]),
            "negative_controls": len(control_ids),
            "operationalizations": len(measurement_ids),
            "analyses": len(analysis_ids),
            "declared_sources": len(source_ids),
        },
        "identifiers": {
            "hypothesis_ids": sorted(hypothesis_ids),
            "estimand_ids": sorted(estimand_ids),
            "prediction_ids": sorted(prediction_ids),
            "measurement_ids": sorted(measurement_ids),
            "analysis_ids": sorted(analysis_ids),
            "control_ids": sorted(control_ids),
        },
        "unresolved_gate_fields": sorted(set(unresolved_gate_names)),
        "notice": (
            "This report validates schema, cross-references, and declared safety "
            "controls only. It does not verify evidence, scientific validity, "
            "novelty, ethics approval, causal identification, or hypothesis merit, "
            "and it does not rank or select candidates."
        ),
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "Validate a bounded local hypothesis JSON record and emit identifiers, "
            "counts, and rule codes without scientific scoring."
        )
    )
    parser.add_argument("record", help="Local hypothesis record JSON")
    parser.add_argument("-o", "--output", help="Optional local JSON report")
    parser.add_argument(
        "--force", action="store_true", help="Replace an existing output file"
    )
    return parser


def main() -> int:
    args = build_parser().parse_args()
    try:
        record = load_hypothesis_record(read_json(args.record))
        report = validate_record(record)
        write_json_report(report, args.output, force=args.force)
        return 0 if report["valid"] else 1
    except ValidationError as exc:
        return error_exit(exc)


if __name__ == "__main__":
    raise SystemExit(main())