diff --git a/src/openamp_foundry/evidence/external_review_packet.py b/src/openamp_foundry/evidence/external_review_packet.py new file mode 100644 index 00000000..3398441d --- /dev/null +++ b/src/openamp_foundry/evidence/external_review_packet.py @@ -0,0 +1,267 @@ +"""External review packet schema — Phase E E1. + +Machine-checkable ERP- schema for complete external review packets. +Enables lab partners and reviewers to verify what a review packet contains +before running experiments. + +Key safety invariants: +- dry_lab_only_attestation must always be True +- proof_ladder_level capped at 2 (computationally nominated) for dry-lab packets +- safety_attestations.no_known_toxicity_claim must be True +- limitations list must be non-empty +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field + + +VALID_CALIBRATION_ASSESSMENTS: frozenset[str] = frozenset({ + "well_calibrated", + "moderately_calibrated", + "poorly_calibrated", + "uninformative", +}) + +DRY_LAB_MAX_PROOF_LADDER_LEVEL: int = 2 +_SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+$") +_GIT_SHA_RE = re.compile(r"^[a-f0-9]{7,40}$") +_AA_SEQ_RE = re.compile(r"^[ACDEFGHIKLMNPQRSTVWY]+$") + + +@dataclass +class CandidateEntry: + candidate_id: str + sequence: str + ensemble_score: float + proof_ladder_level: int + family: str = "" + selection_rationale: str = "" + safety_notes: str = "" + + +@dataclass +class BenchmarkSummary: + auroc: float + benchmark_name: str + n_positives: int + n_negatives: int + + +@dataclass +class CalibrationSummary: + calibration_assessment: str + n_calibration_samples: int = 0 + + +@dataclass +class SafetyAttestations: + reviewed_by_human: bool + safety_gate_passed: bool + no_known_toxicity_claim: bool + hemolysis_assay_mandatory: bool = True + + +@dataclass +class ExternalReviewPacket: + packet_id: str + version: str + generated_at: str + pipeline_version: str + git_sha: str + candidate_count: int + candidates: list[CandidateEntry] = field(default_factory=list) + benchmark_summary: BenchmarkSummary | None = None + calibration_summary: CalibrationSummary | None = None + limitations: list[str] = field(default_factory=list) + safety_attestations: SafetyAttestations | None = None + dry_lab_only_attestation: bool = True + proof_ladder_level: int = 1 + contact: str = "" + notes: str = "" + + +@dataclass +class PacketValidationResult: + is_valid: bool + violations: list[str] + warnings: list[str] + packet_id: str + validation_summary: str + + +def validate_external_review_packet( + packet: ExternalReviewPacket, +) -> PacketValidationResult: + """Validate an ExternalReviewPacket against ERP- schema rules. + + Returns a PacketValidationResult with all violations and warnings. + A packet is valid only when violations is empty. + """ + violations: list[str] = [] + warnings: list[str] = [] + + # Rule 1: packet_id prefix + if not packet.packet_id.startswith("ERP-"): + violations.append( + f"packet_id must start with 'ERP-', got '{packet.packet_id}'" + ) + + # Rule 2: version semver format + if not _SEMVER_RE.match(packet.version): + violations.append( + f"version must be semver (X.Y.Z), got '{packet.version}'" + ) + + # Rule 3: git_sha hex format + if not _GIT_SHA_RE.match(packet.git_sha): + violations.append( + f"git_sha must be 7-40 lowercase hex chars, got '{packet.git_sha}'" + ) + + # Rule 4: candidate_count consistency + if packet.candidate_count != len(packet.candidates): + violations.append( + f"candidate_count ({packet.candidate_count}) does not match " + f"len(candidates) ({len(packet.candidates)})" + ) + + # Rule 5: candidates not empty + if not packet.candidates: + violations.append("candidates must not be empty (minItems: 1)") + + # Rule 6: dry_lab_only_attestation must be True + if not packet.dry_lab_only_attestation: + violations.append( + "dry_lab_only_attestation must be True — " + "this packet contains dry-lab computational outputs only" + ) + + # Rule 7: packet-level proof_ladder_level cap + if packet.proof_ladder_level > DRY_LAB_MAX_PROOF_LADDER_LEVEL: + violations.append( + f"proof_ladder_level ({packet.proof_ladder_level}) exceeds dry-lab cap " + f"of {DRY_LAB_MAX_PROOF_LADDER_LEVEL}. " + "Dry-lab outputs cannot claim beyond 'computationally nominated' (Level 2)." + ) + + # Rule 8: proof_ladder_level must be >= 1 + if packet.proof_ladder_level < 1: + violations.append( + f"proof_ladder_level must be >= 1, got {packet.proof_ladder_level}" + ) + + # Rule 9: safety_attestations + if packet.safety_attestations is None: + violations.append("safety_attestations is required") + else: + if not packet.safety_attestations.no_known_toxicity_claim: + violations.append( + "safety_attestations.no_known_toxicity_claim must be True — " + "dry-lab scores do not prove biological safety" + ) + + # Rule 10: limitations non-empty + if not packet.limitations: + violations.append( + "limitations must contain at least one entry. " + "Minimum: 'Computational outputs are hypotheses and review aids. " + "They are not biological proof.'" + ) + + # Rule 11: benchmark_summary auroc range + if packet.benchmark_summary is not None: + if not (0.0 <= packet.benchmark_summary.auroc <= 1.0): + violations.append( + f"benchmark_summary.auroc must be in [0, 1], " + f"got {packet.benchmark_summary.auroc}" + ) + + # Rule 12: calibration_assessment vocabulary + if packet.calibration_summary is not None: + if ( + packet.calibration_summary.calibration_assessment + not in VALID_CALIBRATION_ASSESSMENTS + ): + violations.append( + f"calibration_summary.calibration_assessment must be one of " + f"{sorted(VALID_CALIBRATION_ASSESSMENTS)}, " + f"got '{packet.calibration_summary.calibration_assessment}'" + ) + + # Rule 13+14: per-candidate validation + for i, cand in enumerate(packet.candidates): + if not (0.0 <= cand.ensemble_score <= 1.0): + violations.append( + f"candidates[{i}].ensemble_score must be in [0, 1], " + f"got {cand.ensemble_score}" + ) + if cand.proof_ladder_level > DRY_LAB_MAX_PROOF_LADDER_LEVEL: + violations.append( + f"candidates[{i}].proof_ladder_level ({cand.proof_ladder_level}) " + f"exceeds dry-lab cap of {DRY_LAB_MAX_PROOF_LADDER_LEVEL}" + ) + + is_valid = len(violations) == 0 + + if is_valid: + summary = ( + f"ERP- packet '{packet.packet_id}' is valid: " + f"{len(packet.candidates)} candidate(s), " + f"proof_ladder_level={packet.proof_ladder_level}, " + "dry_lab_only=True." + ) + else: + summary = ( + f"ERP- packet '{packet.packet_id}' has {len(violations)} violation(s). " + "Fix all violations before sending to external reviewers." + ) + + return PacketValidationResult( + is_valid=is_valid, + violations=violations, + warnings=warnings, + packet_id=packet.packet_id, + validation_summary=summary, + ) + + +def format_external_review_packet(packet: ExternalReviewPacket) -> str: + """Format an ExternalReviewPacket as a human-readable string.""" + result = validate_external_review_packet(packet) + lines = [ + "=== EXTERNAL REVIEW PACKET ===", + f"Packet ID: {packet.packet_id}", + f"Version: {packet.version}", + f"Pipeline: {packet.pipeline_version} ({packet.git_sha})", + f"Generated: {packet.generated_at}", + f"Candidates: {packet.candidate_count}", + f"Proof ladder level: {packet.proof_ladder_level}", + f"Dry-lab only: {'YES' if packet.dry_lab_only_attestation else 'NO'}", + "", + "-- SAFETY ATTESTATIONS --", + ] + if packet.safety_attestations: + lines.extend([ + f" Reviewed by human: {packet.safety_attestations.reviewed_by_human}", + f" Safety gate passed: {packet.safety_attestations.safety_gate_passed}", + f" No toxicity claim: {packet.safety_attestations.no_known_toxicity_claim}", + f" Hemolysis assay mandatory: {packet.safety_attestations.hemolysis_assay_mandatory}", + ]) + lines.extend(["", "-- LIMITATIONS --"]) + for lim in packet.limitations: + lines.append(f" • {lim}") + lines.extend(["", "-- 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}") + lines.extend(["", result.validation_summary]) + if not result.is_valid: + lines.extend([ + "", + "NOTICE: Do not send this packet to external reviewers until all violations are resolved.", + ]) + return "\n".join(lines) diff --git a/tests/evidence/test_external_review_packet.py b/tests/evidence/test_external_review_packet.py new file mode 100644 index 00000000..cc212a9c --- /dev/null +++ b/tests/evidence/test_external_review_packet.py @@ -0,0 +1,401 @@ +"""Tests for external_review_packet module (E1).""" +from __future__ import annotations + +import pytest + +from openamp_foundry.evidence.external_review_packet import ( + BenchmarkSummary, + CalibrationSummary, + CandidateEntry, + DRY_LAB_MAX_PROOF_LADDER_LEVEL, + ExternalReviewPacket, + PacketValidationResult, + SafetyAttestations, + VALID_CALIBRATION_ASSESSMENTS, + format_external_review_packet, + validate_external_review_packet, +) + + +# --- Helpers --- + +def _make_candidate(**kwargs) -> CandidateEntry: + defaults = dict( + candidate_id="AMPF-001", + sequence="KWKLFKKIGAVLKVL", + ensemble_score=0.85, + proof_ladder_level=2, + ) + defaults.update(kwargs) + return CandidateEntry(**defaults) + + +def _make_packet(**kwargs) -> ExternalReviewPacket: + candidate = _make_candidate() + defaults = dict( + packet_id="ERP-2026-001", + version="1.0.0", + generated_at="2026-07-10T00:00:00Z", + pipeline_version="0.9.0", + git_sha="abc1234", + candidate_count=1, + candidates=[candidate], + benchmark_summary=BenchmarkSummary( + auroc=0.82, benchmark_name="500-AMP", n_positives=50, n_negatives=200 + ), + calibration_summary=CalibrationSummary( + calibration_assessment="well_calibrated", n_calibration_samples=100 + ), + limitations=["Computational outputs are hypotheses. Not biological proof."], + safety_attestations=SafetyAttestations( + reviewed_by_human=True, + safety_gate_passed=True, + no_known_toxicity_claim=True, + ), + dry_lab_only_attestation=True, + proof_ladder_level=2, + contact="lab@example.org", + ) + defaults.update(kwargs) + return ExternalReviewPacket(**defaults) + + +# --- CandidateEntry --- + +class TestCandidateEntry: + def test_is_dataclass(self): + c = _make_candidate() + assert isinstance(c, CandidateEntry) + + def test_fields_accessible(self): + c = _make_candidate() + assert c.candidate_id == "AMPF-001" + assert c.ensemble_score == pytest.approx(0.85) + + def test_optional_fields_default(self): + c = _make_candidate() + assert c.family == "" + assert c.selection_rationale == "" + + def test_custom_fields(self): + c = _make_candidate(family="defensin", selection_rationale="high score") + assert c.family == "defensin" + + +# --- BenchmarkSummary --- + +class TestBenchmarkSummary: + def test_is_dataclass(self): + b = BenchmarkSummary(auroc=0.8, benchmark_name="test", n_positives=10, n_negatives=40) + assert isinstance(b, BenchmarkSummary) + + def test_fields_stored(self): + b = BenchmarkSummary(0.75, "500-AMP", 50, 200) + assert b.auroc == pytest.approx(0.75) + assert b.benchmark_name == "500-AMP" + + def test_counts_stored(self): + b = BenchmarkSummary(0.8, "test", 30, 120) + assert b.n_positives == 30 + assert b.n_negatives == 120 + + +# --- CalibrationSummary --- + +class TestCalibrationSummary: + def test_is_dataclass(self): + c = CalibrationSummary(calibration_assessment="well_calibrated") + assert isinstance(c, CalibrationSummary) + + def test_assessment_stored(self): + c = CalibrationSummary("poorly_calibrated", n_calibration_samples=50) + assert c.calibration_assessment == "poorly_calibrated" + + def test_default_samples(self): + c = CalibrationSummary("uninformative") + assert c.n_calibration_samples == 0 + + +# --- SafetyAttestations --- + +class TestSafetyAttestations: + def test_is_dataclass(self): + s = SafetyAttestations(True, True, True) + assert isinstance(s, SafetyAttestations) + + def test_fields_stored(self): + s = SafetyAttestations(reviewed_by_human=True, safety_gate_passed=False, no_known_toxicity_claim=True) + assert s.reviewed_by_human is True + assert s.safety_gate_passed is False + + def test_hemolysis_default_true(self): + s = SafetyAttestations(True, True, True) + assert s.hemolysis_assay_mandatory is True + + +# --- ExternalReviewPacket --- + +class TestExternalReviewPacket: + def test_is_dataclass(self): + p = _make_packet() + assert isinstance(p, ExternalReviewPacket) + + def test_fields_accessible(self): + p = _make_packet() + assert p.packet_id == "ERP-2026-001" + assert p.dry_lab_only_attestation is True + + def test_default_notes_empty(self): + p = _make_packet() + assert p.notes == "" + + def test_candidates_stored(self): + p = _make_packet() + assert len(p.candidates) == 1 + + +# --- PacketValidationResult --- + +class TestPacketValidationResult: + def test_is_dataclass(self): + r = PacketValidationResult(True, [], [], "ERP-001", "ok") + assert isinstance(r, PacketValidationResult) + + def test_fields_accessible(self): + r = PacketValidationResult(False, ["v1"], ["w1"], "ERP-001", "bad") + assert r.is_valid is False + assert len(r.violations) == 1 + assert len(r.warnings) == 1 + + def test_packet_id_stored(self): + r = PacketValidationResult(True, [], [], "ERP-XYZ", "ok") + assert r.packet_id == "ERP-XYZ" + + def test_summary_stored(self): + r = PacketValidationResult(True, [], [], "ERP-001", "all good") + assert r.validation_summary == "all good" + + +# --- validate_external_review_packet --- + +class TestValidateExternalReviewPacket: + def test_valid_packet_passes(self): + result = validate_external_review_packet(_make_packet()) + assert result.is_valid is True + assert result.violations == [] + + def test_returns_validation_result(self): + result = validate_external_review_packet(_make_packet()) + assert isinstance(result, PacketValidationResult) + + def test_invalid_packet_id_prefix(self): + p = _make_packet(packet_id="PKT-001") + result = validate_external_review_packet(p) + assert not result.is_valid + assert any("ERP-" in v for v in result.violations) + + def test_invalid_version_format(self): + p = _make_packet(version="1.0") + result = validate_external_review_packet(p) + assert not result.is_valid + assert any("version" in v for v in result.violations) + + def test_invalid_git_sha_too_short(self): + p = _make_packet(git_sha="abc") + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_invalid_git_sha_uppercase(self): + p = _make_packet(git_sha="ABC1234") + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_candidate_count_mismatch(self): + p = _make_packet(candidate_count=5) + result = validate_external_review_packet(p) + assert not result.is_valid + assert any("candidate_count" in v for v in result.violations) + + def test_empty_candidates(self): + p = _make_packet(candidates=[], candidate_count=0) + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_dry_lab_false_blocked(self): + p = _make_packet(dry_lab_only_attestation=False) + result = validate_external_review_packet(p) + assert not result.is_valid + assert any("dry_lab_only" in v for v in result.violations) + + def test_proof_ladder_above_cap_blocked(self): + p = _make_packet(proof_ladder_level=3) + result = validate_external_review_packet(p) + assert not result.is_valid + assert any("proof_ladder_level" in v or "dry-lab cap" in v for v in result.violations) + + def test_proof_ladder_zero_blocked(self): + p = _make_packet(proof_ladder_level=0) + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_no_toxicity_claim_false_blocked(self): + safety = SafetyAttestations( + reviewed_by_human=True, + safety_gate_passed=True, + no_known_toxicity_claim=False, + ) + p = _make_packet(safety_attestations=safety) + result = validate_external_review_packet(p) + assert not result.is_valid + assert any("toxicity" in v for v in result.violations) + + def test_none_safety_attestations_blocked(self): + p = _make_packet(safety_attestations=None) + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_empty_limitations_blocked(self): + p = _make_packet(limitations=[]) + result = validate_external_review_packet(p) + assert not result.is_valid + assert any("limitations" in v for v in result.violations) + + def test_auroc_above_one_blocked(self): + bs = BenchmarkSummary(auroc=1.1, benchmark_name="test", n_positives=10, n_negatives=40) + p = _make_packet(benchmark_summary=bs) + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_auroc_below_zero_blocked(self): + bs = BenchmarkSummary(auroc=-0.1, benchmark_name="test", n_positives=10, n_negatives=40) + p = _make_packet(benchmark_summary=bs) + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_invalid_calibration_assessment_blocked(self): + cs = CalibrationSummary(calibration_assessment="excellent") + p = _make_packet(calibration_summary=cs) + result = validate_external_review_packet(p) + assert not result.is_valid + assert any("calibration_assessment" in v for v in result.violations) + + def test_valid_calibration_assessments_accepted(self): + for assessment in VALID_CALIBRATION_ASSESSMENTS: + cs = CalibrationSummary(calibration_assessment=assessment) + p = _make_packet(calibration_summary=cs) + result = validate_external_review_packet(p) + cal_violations = [v for v in result.violations if "calibration_assessment" in v] + assert cal_violations == [], f"{assessment} should be valid" + + def test_candidate_ensemble_score_above_one_blocked(self): + c = _make_candidate(ensemble_score=1.5) + p = _make_packet(candidates=[c], candidate_count=1) + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_candidate_ensemble_score_below_zero_blocked(self): + c = _make_candidate(ensemble_score=-0.1) + p = _make_packet(candidates=[c], candidate_count=1) + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_candidate_proof_ladder_above_cap_blocked(self): + c = _make_candidate(proof_ladder_level=3) + p = _make_packet(candidates=[c], candidate_count=1) + result = validate_external_review_packet(p) + assert not result.is_valid + + def test_candidate_proof_ladder_at_cap_ok(self): + c = _make_candidate(proof_ladder_level=DRY_LAB_MAX_PROOF_LADDER_LEVEL) + p = _make_packet(candidates=[c], candidate_count=1) + result = validate_external_review_packet(p) + candidate_violations = [v for v in result.violations if "candidates[0]" in v] + assert candidate_violations == [] + + def test_valid_summary_contains_packet_id(self): + result = validate_external_review_packet(_make_packet()) + assert "ERP-2026-001" in result.validation_summary + + def test_violation_summary_contains_count(self): + p = _make_packet(packet_id="NOT-VALID", dry_lab_only_attestation=False) + result = validate_external_review_packet(p) + assert str(len(result.violations)) in result.validation_summary or "violation" in result.validation_summary + + def test_dry_lab_max_level_is_2(self): + assert DRY_LAB_MAX_PROOF_LADDER_LEVEL == 2 + + def test_valid_calibration_vocabulary_size(self): + assert len(VALID_CALIBRATION_ASSESSMENTS) == 4 + + def test_multiple_violations_all_reported(self): + p = _make_packet( + packet_id="BAD", + dry_lab_only_attestation=False, + limitations=[], + ) + result = validate_external_review_packet(p) + assert len(result.violations) >= 3 + + def test_git_sha_40_chars_valid(self): + sha = "a" * 40 + p = _make_packet(git_sha=sha) + result = validate_external_review_packet(p) + sha_violations = [v for v in result.violations if "git_sha" in v] + assert sha_violations == [] + + def test_git_sha_7_chars_valid(self): + p = _make_packet(git_sha="abcdef0") + result = validate_external_review_packet(p) + sha_violations = [v for v in result.violations if "git_sha" in v] + assert sha_violations == [] + + +# --- format_external_review_packet --- + +class TestFormatExternalReviewPacket: + def test_returns_string(self): + assert isinstance(format_external_review_packet(_make_packet()), str) + + def test_contains_header(self): + assert "EXTERNAL REVIEW PACKET" in format_external_review_packet(_make_packet()) + + def test_contains_packet_id(self): + assert "ERP-2026-001" in format_external_review_packet(_make_packet()) + + def test_valid_packet_shows_valid(self): + assert "VALID" in format_external_review_packet(_make_packet()) + + def test_invalid_packet_shows_invalid(self): + p = _make_packet(dry_lab_only_attestation=False) + assert "INVALID" in format_external_review_packet(p) + + def test_contains_limitations(self): + text = format_external_review_packet(_make_packet()) + assert "LIMITATIONS" in text + + def test_contains_safety_section(self): + text = format_external_review_packet(_make_packet()) + assert "SAFETY" in text + + def test_dry_lab_yes_shown(self): + text = format_external_review_packet(_make_packet()) + assert "YES" in text + + def test_notice_in_invalid_output(self): + p = _make_packet(dry_lab_only_attestation=False) + text = format_external_review_packet(p) + assert "NOTICE" in text + + def test_candidate_count_shown(self): + text = format_external_review_packet(_make_packet()) + assert "Candidates" in text + + def test_pipeline_version_shown(self): + text = format_external_review_packet(_make_packet()) + assert "0.9.0" in text + + def test_summary_in_output(self): + p = _make_packet() + result = validate_external_review_packet(p) + text = format_external_review_packet(p) + assert result.validation_summary in text diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index 01be7764..aa83b666 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -3,7 +3,7 @@ import sys -BASELINE = 7339 +BASELINE = 7402 TOLERANCE = 0.05