From 05dfcf56a4e745367a028685ad590210bdf04da7 Mon Sep 17 00:00:00 2001 From: OpenCode Date: Fri, 10 Jul 2026 11:04:02 +0700 Subject: [PATCH] feat: Phase E E5 pilot pre-registration schema -- PRR- record freezes selection thresholds before wet-lab, prevents post-hoc adjustment, 63 tests (#896) --- .../evidence/pilot_preregistration.py | 258 +++++++++++++ tests/evidence/test_pilot_preregistration.py | 348 ++++++++++++++++++ tests/test_test_count_regression.py | 2 +- 3 files changed, 607 insertions(+), 1 deletion(-) create mode 100644 src/openamp_foundry/evidence/pilot_preregistration.py create mode 100644 tests/evidence/test_pilot_preregistration.py diff --git a/src/openamp_foundry/evidence/pilot_preregistration.py b/src/openamp_foundry/evidence/pilot_preregistration.py new file mode 100644 index 00000000..49fc69ca --- /dev/null +++ b/src/openamp_foundry/evidence/pilot_preregistration.py @@ -0,0 +1,258 @@ +"""Pilot pre-registration schema — Phase E E5. + +PRR- records freeze the candidate selection criteria, score thresholds, and +primary hypothesis BEFORE any wet-lab experiment begins. This prevents +post-hoc threshold adjustment to match results — the most common form of +silent overclaiming in computational-to-experimental pipelines. + +A locked PRR- record is the pre-registration contract between the dry-lab +pipeline and the wet-lab team. +""" +from __future__ import annotations + +from dataclasses import dataclass, field + + +VALID_OUTCOME_METRICS: frozenset[str] = frozenset({ + "mic_reduction", + "zone_of_inhibition", + "minimum_inhibitory_concentration", + "percent_viability", + "auc_growth_curve", + "colony_forming_units", + "optical_density", + "time_to_kill", +}) + +VALID_AMENDMENT_REASONS: frozenset[str] = frozenset({ + "clerical_error_correction", + "scope_reduction", + "control_substitution", + "metric_clarification", + "timeline_extension", +}) + + +@dataclass +class ScoreThreshold: + score_name: str + threshold_value: float + direction: str + + +@dataclass +class PilotPreregistration: + record_id: str + version: str + frozen_at: str + pipeline_version: str + git_sha: str + primary_hypothesis: str + selection_criteria: list[str] = field(default_factory=list) + score_thresholds: list[ScoreThreshold] = field(default_factory=list) + n_candidates_planned: int = 0 + positive_control: str = "" + negative_control: str = "" + outcome_metric: str = "" + dry_lab_only_declaration: bool = True + is_locked: bool = False + amendment_count: int = 0 + amendment_reasons: list[str] = field(default_factory=list) + notes: str = "" + + +@dataclass +class PreregistrationValidationResult: + is_valid: bool + violations: list[str] + warnings: list[str] + record_id: str + validation_summary: str + + +_SEMVER_RE = __import__("re").compile(r"^\d+\.\d+\.\d+$") +_GIT_SHA_RE = __import__("re").compile(r"^[a-f0-9]{7,40}$") + + +def validate_pilot_preregistration( + record: PilotPreregistration, +) -> PreregistrationValidationResult: + """Validate a PilotPreregistration against PRR- schema rules. + + Returns a PreregistrationValidationResult with all violations. + A record is valid only when violations is empty. + """ + violations: list[str] = [] + warnings: list[str] = [] + + # Rule 1: record_id prefix + if not record.record_id.startswith("PRR-"): + violations.append( + f"record_id must start with 'PRR-', got '{record.record_id}'" + ) + + # Rule 2: version semver + if not _SEMVER_RE.match(record.version): + violations.append( + f"version must be semver (X.Y.Z), got '{record.version}'" + ) + + # Rule 3: git_sha hex + if not _GIT_SHA_RE.match(record.git_sha): + violations.append( + f"git_sha must be 7-40 lowercase hex chars, got '{record.git_sha}'" + ) + + # Rule 4: dry_lab_only_declaration must be True + if not record.dry_lab_only_declaration: + violations.append( + "dry_lab_only_declaration must be True — " + "pre-registration captures dry-lab selection logic only" + ) + + # Rule 5: selection_criteria non-empty + if not record.selection_criteria: + violations.append( + "selection_criteria must contain at least one criterion" + ) + + # Rule 6: score_thresholds non-empty + if not record.score_thresholds: + violations.append( + "score_thresholds must contain at least one threshold — " + "thresholds must be declared before the experiment" + ) + + # Rule 7: n_candidates_planned >= 1 + if record.n_candidates_planned < 1: + violations.append( + f"n_candidates_planned must be >= 1, got {record.n_candidates_planned}" + ) + + # Rule 8: outcome_metric vocabulary + if record.outcome_metric not in VALID_OUTCOME_METRICS: + violations.append( + f"outcome_metric must be one of {sorted(VALID_OUTCOME_METRICS)}, " + f"got '{record.outcome_metric}'" + ) + + # Rule 9: primary_hypothesis not empty + if not record.primary_hypothesis.strip(): + violations.append("primary_hypothesis must not be empty") + + # Rule 10: positive_control not empty + if not record.positive_control.strip(): + violations.append( + "positive_control must not be empty — " + "a pre-registered experiment without a positive control cannot be interpreted" + ) + + # Rule 11: negative_control not empty + if not record.negative_control.strip(): + violations.append( + "negative_control must not be empty — " + "a pre-registered experiment without a negative control cannot be interpreted" + ) + + # Rule 12: frozen_at not empty + if not record.frozen_at.strip(): + violations.append("frozen_at must not be empty") + + # Rule 13: threshold values in [0, 1] + for t in record.score_thresholds: + if not (0.0 <= t.threshold_value <= 1.0): + violations.append( + f"score_thresholds[{t.score_name}].threshold_value must be in [0, 1], " + f"got {t.threshold_value}" + ) + + # Rule 14: amendment_reasons vocabulary + for reason in record.amendment_reasons: + if reason not in VALID_AMENDMENT_REASONS: + violations.append( + f"amendment_reasons contains invalid entry '{reason}'. " + f"Valid reasons: {sorted(VALID_AMENDMENT_REASONS)}" + ) + + # Warning: locked record with amendments + if record.is_locked and record.amendment_count > 0: + warnings.append( + f"Record is locked with {record.amendment_count} amendment(s). " + "Amendments after lock require human review." + ) + + is_valid = len(violations) == 0 + + if is_valid: + summary = ( + f"PRR- record '{record.record_id}' is valid: " + f"{record.n_candidates_planned} candidate(s) planned, " + f"outcome_metric='{record.outcome_metric}', " + f"locked={record.is_locked}." + ) + else: + summary = ( + f"PRR- record '{record.record_id}' has {len(violations)} violation(s). " + "Fix all violations before beginning any wet-lab experiment." + ) + + return PreregistrationValidationResult( + is_valid=is_valid, + violations=violations, + warnings=warnings, + record_id=record.record_id, + validation_summary=summary, + ) + + +def format_pilot_preregistration(record: PilotPreregistration) -> str: + """Format a PilotPreregistration as a human-readable string.""" + result = validate_pilot_preregistration(record) + lines = [ + "=== PILOT PRE-REGISTRATION RECORD ===", + f"Record ID: {record.record_id}", + f"Version: {record.version}", + f"Frozen at: {record.frozen_at}", + f"Pipeline: {record.pipeline_version} ({record.git_sha})", + f"Locked: {'YES' if record.is_locked else 'NO'}", + f"Amendment count: {record.amendment_count}", + "", + "-- HYPOTHESIS --", + f" {record.primary_hypothesis}", + "", + "-- SELECTION CRITERIA --", + ] + for criterion in record.selection_criteria: + lines.append(f" • {criterion}") + lines.extend(["", "-- SCORE THRESHOLDS (frozen) --"]) + for t in record.score_thresholds: + lines.append(f" {t.score_name}: {t.threshold_value:.4f} ({t.direction})") + lines.extend([ + "", + "-- EXPERIMENT DESIGN --", + f" Candidates planned: {record.n_candidates_planned}", + f" Outcome metric: {record.outcome_metric}", + f" Positive control: {record.positive_control}", + f" Negative control: {record.negative_control}", + f" Dry-lab only declaration: {'YES' if record.dry_lab_only_declaration else 'NO'}", + "", + "-- VALIDATION --", + ]) + if result.is_valid: + lines.append(" STATUS: VALID") + else: + lines.append(f" STATUS: INVALID ({len(result.violations)} violation(s))") + for v in result.violations: + lines.append(f" ✗ {v}") + if result.warnings: + lines.append("") + for w in result.warnings: + lines.append(f" ⚠ {w}") + lines.extend(["", result.validation_summary]) + if not result.is_valid: + lines.extend([ + "", + "NOTICE: Do not begin any wet-lab experiment until all violations are resolved.", + "A valid, locked PRR- record is required before experiment start.", + ]) + return "\n".join(lines) diff --git a/tests/evidence/test_pilot_preregistration.py b/tests/evidence/test_pilot_preregistration.py new file mode 100644 index 00000000..9c942f03 --- /dev/null +++ b/tests/evidence/test_pilot_preregistration.py @@ -0,0 +1,348 @@ +"""Tests for pilot_preregistration module (E5).""" +from __future__ import annotations + +import pytest + +from openamp_foundry.evidence.pilot_preregistration import ( + PilotPreregistration, + PreregistrationValidationResult, + ScoreThreshold, + VALID_AMENDMENT_REASONS, + VALID_OUTCOME_METRICS, + format_pilot_preregistration, + validate_pilot_preregistration, +) + + +# --- Helpers --- + +def _make_threshold(**kwargs) -> ScoreThreshold: + defaults = dict(score_name="ensemble_score", threshold_value=0.75, direction="above") + defaults.update(kwargs) + return ScoreThreshold(**defaults) + + +def _make_record(**kwargs) -> PilotPreregistration: + defaults = dict( + record_id="PRR-2026-001", + version="1.0.0", + frozen_at="2026-07-10T00:00:00Z", + pipeline_version="0.9.0", + git_sha="abc1234", + primary_hypothesis="High-scoring candidates will show MIC reduction vs. control.", + selection_criteria=["ensemble_score >= 0.75", "toxicity_score < 0.3"], + score_thresholds=[_make_threshold()], + n_candidates_planned=5, + positive_control="colistin_1ug_ml", + negative_control="pbs_vehicle", + outcome_metric="minimum_inhibitory_concentration", + dry_lab_only_declaration=True, + is_locked=False, + ) + defaults.update(kwargs) + return PilotPreregistration(**defaults) + + +# --- ScoreThreshold --- + +class TestScoreThreshold: + def test_is_dataclass(self): + t = _make_threshold() + assert isinstance(t, ScoreThreshold) + + def test_fields_accessible(self): + t = ScoreThreshold("activity_score", 0.80, "above") + assert t.score_name == "activity_score" + assert t.threshold_value == pytest.approx(0.80) + assert t.direction == "above" + + def test_threshold_stored(self): + t = _make_threshold(threshold_value=0.5) + assert t.threshold_value == pytest.approx(0.5) + + +# --- PilotPreregistration --- + +class TestPilotPreregistration: + def test_is_dataclass(self): + r = _make_record() + assert isinstance(r, PilotPreregistration) + + def test_fields_accessible(self): + r = _make_record() + assert r.record_id == "PRR-2026-001" + assert r.dry_lab_only_declaration is True + + def test_default_notes_empty(self): + r = _make_record() + assert r.notes == "" + + def test_default_amendment_count(self): + r = _make_record() + assert r.amendment_count == 0 + + def test_is_locked_default_false(self): + r = _make_record() + assert r.is_locked is False + + +# --- PreregistrationValidationResult --- + +class TestPreregistrationValidationResult: + def test_is_dataclass(self): + r = PreregistrationValidationResult(True, [], [], "PRR-001", "ok") + assert isinstance(r, PreregistrationValidationResult) + + def test_fields_accessible(self): + r = PreregistrationValidationResult(False, ["v1"], ["w1"], "PRR-001", "bad") + assert r.is_valid is False + assert len(r.violations) == 1 + assert r.record_id == "PRR-001" + + def test_summary_stored(self): + r = PreregistrationValidationResult(True, [], [], "PRR-001", "all good") + assert r.validation_summary == "all good" + + +# --- validate_pilot_preregistration --- + +class TestValidatePilotPreregistration: + def test_valid_record_passes(self): + result = validate_pilot_preregistration(_make_record()) + assert result.is_valid is True + assert result.violations == [] + + def test_returns_validation_result(self): + result = validate_pilot_preregistration(_make_record()) + assert isinstance(result, PreregistrationValidationResult) + + def test_invalid_record_id_prefix(self): + r = _make_record(record_id="REC-001") + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("PRR-" in v for v in result.violations) + + def test_invalid_version_format(self): + r = _make_record(version="v1.0") + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("version" in v for v in result.violations) + + def test_invalid_git_sha_too_short(self): + r = _make_record(git_sha="abc") + result = validate_pilot_preregistration(r) + assert not result.is_valid + + def test_invalid_git_sha_uppercase(self): + r = _make_record(git_sha="ABC1234") + result = validate_pilot_preregistration(r) + assert not result.is_valid + + def test_dry_lab_false_blocked(self): + r = _make_record(dry_lab_only_declaration=False) + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("dry_lab_only" in v for v in result.violations) + + def test_empty_selection_criteria_blocked(self): + r = _make_record(selection_criteria=[]) + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("selection_criteria" in v for v in result.violations) + + def test_empty_score_thresholds_blocked(self): + r = _make_record(score_thresholds=[]) + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("score_thresholds" in v for v in result.violations) + + def test_zero_candidates_blocked(self): + r = _make_record(n_candidates_planned=0) + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("n_candidates_planned" in v for v in result.violations) + + def test_negative_candidates_blocked(self): + r = _make_record(n_candidates_planned=-1) + result = validate_pilot_preregistration(r) + assert not result.is_valid + + def test_invalid_outcome_metric_blocked(self): + r = _make_record(outcome_metric="made_up_metric") + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("outcome_metric" in v for v in result.violations) + + def test_valid_outcome_metrics_accepted(self): + for metric in VALID_OUTCOME_METRICS: + r = _make_record(outcome_metric=metric) + result = validate_pilot_preregistration(r) + metric_violations = [v for v in result.violations if "outcome_metric" in v] + assert metric_violations == [], f"{metric} should be valid" + + def test_empty_hypothesis_blocked(self): + r = _make_record(primary_hypothesis="") + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("primary_hypothesis" in v for v in result.violations) + + def test_whitespace_only_hypothesis_blocked(self): + r = _make_record(primary_hypothesis=" ") + result = validate_pilot_preregistration(r) + assert not result.is_valid + + def test_empty_positive_control_blocked(self): + r = _make_record(positive_control="") + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("positive_control" in v for v in result.violations) + + def test_empty_negative_control_blocked(self): + r = _make_record(negative_control="") + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("negative_control" in v for v in result.violations) + + def test_empty_frozen_at_blocked(self): + r = _make_record(frozen_at="") + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("frozen_at" in v for v in result.violations) + + def test_threshold_above_one_blocked(self): + t = _make_threshold(threshold_value=1.5) + r = _make_record(score_thresholds=[t]) + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("threshold_value" in v for v in result.violations) + + def test_threshold_below_zero_blocked(self): + t = _make_threshold(threshold_value=-0.1) + r = _make_record(score_thresholds=[t]) + result = validate_pilot_preregistration(r) + assert not result.is_valid + + def test_threshold_at_zero_valid(self): + t = _make_threshold(threshold_value=0.0) + r = _make_record(score_thresholds=[t]) + result = validate_pilot_preregistration(r) + threshold_violations = [v for v in result.violations if "threshold_value" in v] + assert threshold_violations == [] + + def test_threshold_at_one_valid(self): + t = _make_threshold(threshold_value=1.0) + r = _make_record(score_thresholds=[t]) + result = validate_pilot_preregistration(r) + threshold_violations = [v for v in result.violations if "threshold_value" in v] + assert threshold_violations == [] + + def test_invalid_amendment_reason_blocked(self): + r = _make_record(amendment_reasons=["add_new_outcome"]) + result = validate_pilot_preregistration(r) + assert not result.is_valid + assert any("amendment_reasons" in v for v in result.violations) + + def test_valid_amendment_reason_accepted(self): + for reason in VALID_AMENDMENT_REASONS: + r = _make_record(amendment_reasons=[reason]) + result = validate_pilot_preregistration(r) + reason_violations = [v for v in result.violations if "amendment_reasons" in v] + assert reason_violations == [], f"{reason} should be valid" + + def test_locked_with_amendments_produces_warning(self): + r = _make_record(is_locked=True, amendment_count=2) + result = validate_pilot_preregistration(r) + assert len(result.warnings) >= 1 + assert any("amendment" in w for w in result.warnings) + + def test_valid_summary_contains_record_id(self): + result = validate_pilot_preregistration(_make_record()) + assert "PRR-2026-001" in result.validation_summary + + def test_violation_summary_mentions_violations(self): + r = _make_record(record_id="BAD", dry_lab_only_declaration=False) + result = validate_pilot_preregistration(r) + assert "violation" in result.validation_summary + + def test_valid_outcome_metrics_set_size(self): + assert len(VALID_OUTCOME_METRICS) == 8 + + def test_valid_amendment_reasons_set_size(self): + assert len(VALID_AMENDMENT_REASONS) == 5 + + def test_multiple_violations_all_reported(self): + r = _make_record( + record_id="BAD", + dry_lab_only_declaration=False, + selection_criteria=[], + score_thresholds=[], + ) + result = validate_pilot_preregistration(r) + assert len(result.violations) >= 4 + + def test_locked_record_no_violations(self): + r = _make_record(is_locked=True) + result = validate_pilot_preregistration(r) + assert result.is_valid + + def test_multiple_thresholds_accepted(self): + thresholds = [ + _make_threshold(score_name="ensemble", threshold_value=0.75), + _make_threshold(score_name="activity", threshold_value=0.70), + _make_threshold(score_name="safety", threshold_value=0.80), + ] + r = _make_record(score_thresholds=thresholds) + result = validate_pilot_preregistration(r) + assert result.is_valid + + +# --- format_pilot_preregistration --- + +class TestFormatPilotPreregistration: + def test_returns_string(self): + assert isinstance(format_pilot_preregistration(_make_record()), str) + + def test_contains_header(self): + assert "PILOT PRE-REGISTRATION" in format_pilot_preregistration(_make_record()) + + def test_contains_record_id(self): + assert "PRR-2026-001" in format_pilot_preregistration(_make_record()) + + def test_valid_record_shows_valid(self): + assert "VALID" in format_pilot_preregistration(_make_record()) + + def test_invalid_record_shows_invalid(self): + r = _make_record(dry_lab_only_declaration=False) + assert "INVALID" in format_pilot_preregistration(r) + + def test_contains_hypothesis(self): + text = format_pilot_preregistration(_make_record()) + assert "HYPOTHESIS" in text + + def test_contains_thresholds_section(self): + text = format_pilot_preregistration(_make_record()) + assert "THRESHOLD" in text + + def test_contains_outcome_metric(self): + text = format_pilot_preregistration(_make_record()) + assert "minimum_inhibitory_concentration" in text + + def test_notice_in_invalid_output(self): + r = _make_record(dry_lab_only_declaration=False) + text = format_pilot_preregistration(r) + assert "NOTICE" in text + + def test_positive_control_shown(self): + text = format_pilot_preregistration(_make_record()) + assert "colistin_1ug_ml" in text + + def test_summary_in_output(self): + r = _make_record() + result = validate_pilot_preregistration(r) + text = format_pilot_preregistration(r) + assert result.validation_summary in text + + def test_locked_status_shown(self): + r = _make_record(is_locked=True) + text = format_pilot_preregistration(r) + assert "YES" in text diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index aa83b666..ee7379bc 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -3,7 +3,7 @@ import sys -BASELINE = 7402 +BASELINE = 7465 TOLERANCE = 0.05