From c1ece986a9fef909e18bda625acb3a532d237c9c Mon Sep 17 00:00:00 2001 From: OpenCode Date: Fri, 10 Jul 2026 13:31:06 +0700 Subject: [PATCH] feat: Phase Q Q3 post-experiment calibration update schema (PCU-) -- dry_lab_only=False enforced, human_reviewed=True required, VALID_UPDATE_DIRECTIONS (5), VALID_CALIBRATION_TRIGGER_TYPES (7), n_whr_records consistency check; 63 tests (#949) --- .../post_experiment_calibration_update.py | 248 +++++++++++++ ...test_post_experiment_calibration_update.py | 343 ++++++++++++++++++ tests/test_test_count_regression.py | 2 +- 3 files changed, 592 insertions(+), 1 deletion(-) create mode 100644 src/openamp_foundry/evidence/post_experiment_calibration_update.py create mode 100644 tests/evidence/test_post_experiment_calibration_update.py diff --git a/src/openamp_foundry/evidence/post_experiment_calibration_update.py b/src/openamp_foundry/evidence/post_experiment_calibration_update.py new file mode 100644 index 00000000..69167385 --- /dev/null +++ b/src/openamp_foundry/evidence/post_experiment_calibration_update.py @@ -0,0 +1,248 @@ +"""Post-experiment calibration update schema — Phase Q Q3. + +PCU- (Post-Experiment Calibration Update): machine-readable record of how +wet-lab experimental results inform a calibration update decision. + +Key safety invariants: +- dry_lab_only MUST be False (these are real-experiment-triggered updates) +- n_whr_records must be >= 1 (at least one WHR- record must justify the update) +- update_direction must be one of: "increase_threshold", "decrease_threshold", + "no_change", "expand_candidate_pool", "restrict_candidate_pool" +- update_magnitude must be in [0.0, 1.0] +- rationale must be non-empty +- limitations must be non-empty +- human_reviewed must be True (calibration updates require human review sign-off) +""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass, field + + +VALID_UPDATE_DIRECTIONS: frozenset[str] = frozenset({ + "increase_threshold", + "decrease_threshold", + "no_change", + "expand_candidate_pool", + "restrict_candidate_pool", +}) + +VALID_CALIBRATION_TRIGGER_TYPES: frozenset[str] = frozenset({ + "hit_rate_above_baseline", + "hit_rate_below_baseline", + "novel_family_detected", + "safety_signal_observed", + "systematic_prediction_error", + "insufficient_data", + "scheduled_review", +}) + +_PCU_RE = re.compile(r"^PCU-[A-Za-z0-9_-]+$") +MAX_RATIONALE_LEN = 1000 +MAX_NOTES_LEN = 500 + + +@dataclass +class PostExperimentCalibrationUpdate: + """Machine-readable record of a post-experiment calibration update decision. + + human_reviewed must be True — calibration updates require human oversight. + dry_lab_only must be False — this is triggered by real experimental results. + """ + pcu_id: str + update_date: str + n_whr_records: int + whr_ids: list[str] + trigger_type: str + update_direction: str + update_magnitude: float + rationale: str + limitations: list[str] + human_reviewed: bool + dry_lab_only: bool + reviewer_id: str = "" + related_ccs_id: str = "" + notes: str = "" + prediction_error_before: float = 0.0 + prediction_error_after: float = 0.0 + + +@dataclass +class PCUValidationResult: + valid: bool + violations: list[str] + pcu_id: str | None + + +def validate_post_experiment_calibration_update( + record: PostExperimentCalibrationUpdate, +) -> PCUValidationResult: + """Validate a PostExperimentCalibrationUpdate and return violations.""" + violations: list[str] = [] + + # ID format + if not _PCU_RE.match(record.pcu_id): + violations.append( + f"pcu_id '{record.pcu_id}' must match PCU-[A-Za-z0-9_-]+" + ) + + # dry_lab_only must be False + if record.dry_lab_only is True: + violations.append( + "dry_lab_only must be False: " + "PCU- records are triggered by real experimental results" + ) + + # human_reviewed must be True + if record.human_reviewed is not True: + violations.append( + "human_reviewed must be True: " + "calibration updates require human review sign-off" + ) + + # n_whr_records must be >= 1 + if record.n_whr_records < 1: + violations.append( + f"n_whr_records {record.n_whr_records} must be >= 1: " + "at least one WHR- record must justify the update" + ) + + # whr_ids must be non-empty and consistent with n_whr_records + if not record.whr_ids: + violations.append("whr_ids must be non-empty") + elif len(record.whr_ids) != record.n_whr_records: + violations.append( + f"whr_ids length {len(record.whr_ids)} must equal " + f"n_whr_records {record.n_whr_records}" + ) + + # update_date must be non-empty + if not record.update_date: + violations.append("update_date must be non-empty") + + # trigger_type must be valid + if record.trigger_type not in VALID_CALIBRATION_TRIGGER_TYPES: + violations.append( + f"trigger_type '{record.trigger_type}' is not in " + f"VALID_CALIBRATION_TRIGGER_TYPES" + ) + + # update_direction must be valid + if record.update_direction not in VALID_UPDATE_DIRECTIONS: + violations.append( + f"update_direction '{record.update_direction}' is not in " + f"VALID_UPDATE_DIRECTIONS" + ) + + # update_magnitude must be in [0.0, 1.0] + if not math.isfinite(record.update_magnitude): + violations.append( + f"update_magnitude {record.update_magnitude} must be finite" + ) + elif not (0.0 <= record.update_magnitude <= 1.0): + violations.append( + f"update_magnitude {record.update_magnitude} must be in [0.0, 1.0]" + ) + + # rationale must be non-empty and within length limit + if not record.rationale: + violations.append("rationale must be non-empty") + elif len(record.rationale) > MAX_RATIONALE_LEN: + violations.append( + f"rationale length {len(record.rationale)} exceeds " + f"maximum {MAX_RATIONALE_LEN}" + ) + + # limitations must be non-empty + if not record.limitations: + violations.append( + "limitations must be non-empty: " + "every calibration update has scope and data limitations" + ) + + # prediction_error values must be finite and non-negative + for field_name, val in [ + ("prediction_error_before", record.prediction_error_before), + ("prediction_error_after", record.prediction_error_after), + ]: + if not math.isfinite(val): + violations.append(f"{field_name} {val} must be finite") + elif val < 0.0: + violations.append(f"{field_name} {val} must be non-negative") + + # notes length cap + if len(record.notes) > MAX_NOTES_LEN: + violations.append( + f"notes length {len(record.notes)} exceeds maximum {MAX_NOTES_LEN}" + ) + + return PCUValidationResult( + valid=len(violations) == 0, + violations=violations, + pcu_id=record.pcu_id if _PCU_RE.match(record.pcu_id) else None, + ) + + +def build_post_experiment_calibration_update( + pcu_id: str, + update_date: str, + whr_ids: list[str], + trigger_type: str, + update_direction: str, + update_magnitude: float, + rationale: str, + limitations: list[str], + reviewer_id: str = "", + related_ccs_id: str = "", + notes: str = "", + prediction_error_before: float = 0.0, + prediction_error_after: float = 0.0, +) -> PostExperimentCalibrationUpdate: + """Build a PCU record with dry_lab_only=False and human_reviewed=True enforced.""" + return PostExperimentCalibrationUpdate( + pcu_id=pcu_id, + update_date=update_date, + n_whr_records=len(whr_ids), + whr_ids=whr_ids, + trigger_type=trigger_type, + update_direction=update_direction, + update_magnitude=update_magnitude, + rationale=rationale, + limitations=limitations, + human_reviewed=True, + dry_lab_only=False, + reviewer_id=reviewer_id, + related_ccs_id=related_ccs_id, + notes=notes, + prediction_error_before=prediction_error_before, + prediction_error_after=prediction_error_after, + ) + + +def format_post_experiment_calibration_update( + record: PostExperimentCalibrationUpdate, +) -> str: + """Return a human-readable summary of the calibration update record.""" + result = validate_post_experiment_calibration_update(record) + lines = [ + f"PCU Record: {record.pcu_id}", + f" Date: {record.update_date}", + f" WHR records: {record.n_whr_records} ({', '.join(record.whr_ids[:3])}{'...' if len(record.whr_ids) > 3 else ''})", + f" Trigger: {record.trigger_type}", + f" Direction: {record.update_direction} (magnitude: {record.update_magnitude:.3f})", + f" human_reviewed: {record.human_reviewed}", + f" dry_lab_only: {record.dry_lab_only}", + f" Rationale: {record.rationale[:100]}{'...' if len(record.rationale) > 100 else ''}", + f" Limitations ({len(record.limitations)}):", + ] + for lim in record.limitations: + lines.append(f" - {lim}") + if result.valid: + lines.append(" Status: VALID") + else: + lines.append(f" Status: INVALID ({len(result.violations)} violation(s))") + for v in result.violations: + lines.append(f" VIOLATION: {v}") + return "\n".join(lines) diff --git a/tests/evidence/test_post_experiment_calibration_update.py b/tests/evidence/test_post_experiment_calibration_update.py new file mode 100644 index 00000000..d0b7728c --- /dev/null +++ b/tests/evidence/test_post_experiment_calibration_update.py @@ -0,0 +1,343 @@ +"""Tests for Phase Q Q3 post-experiment calibration update schema (PCU-).""" + +from __future__ import annotations + +import math + +import pytest + +from openamp_foundry.evidence.post_experiment_calibration_update import ( + VALID_CALIBRATION_TRIGGER_TYPES, + VALID_UPDATE_DIRECTIONS, + MAX_NOTES_LEN, + MAX_RATIONALE_LEN, + PostExperimentCalibrationUpdate, + PCUValidationResult, + build_post_experiment_calibration_update, + format_post_experiment_calibration_update, + validate_post_experiment_calibration_update, +) + + +def _make_valid_record(**overrides) -> PostExperimentCalibrationUpdate: + defaults = dict( + pcu_id="PCU-0001", + update_date="2026-07-10", + n_whr_records=2, + whr_ids=["WHR-0001", "WHR-0002"], + trigger_type="hit_rate_above_baseline", + update_direction="expand_candidate_pool", + update_magnitude=0.15, + rationale="Hit rate exceeded baseline by 2x in first pilot batch.", + limitations=[ + "Based on single pilot batch only.", + "E. coli ATCC 25922 only; no clinical isolates tested.", + ], + human_reviewed=True, + dry_lab_only=False, + reviewer_id="LAB-REVIEWER-001", + related_ccs_id="CCS-0001", + notes="", + prediction_error_before=0.18, + prediction_error_after=0.12, + ) + defaults.update(overrides) + return PostExperimentCalibrationUpdate(**defaults) + + +class TestPCUConstants: + def test_valid_update_directions_is_frozenset(self): + assert isinstance(VALID_UPDATE_DIRECTIONS, frozenset) + + def test_increase_threshold_in_directions(self): + assert "increase_threshold" in VALID_UPDATE_DIRECTIONS + + def test_no_change_in_directions(self): + assert "no_change" in VALID_UPDATE_DIRECTIONS + + def test_valid_trigger_types_is_frozenset(self): + assert isinstance(VALID_CALIBRATION_TRIGGER_TYPES, frozenset) + + def test_hit_rate_above_baseline_in_triggers(self): + assert "hit_rate_above_baseline" in VALID_CALIBRATION_TRIGGER_TYPES + + def test_insufficient_data_in_triggers(self): + assert "insufficient_data" in VALID_CALIBRATION_TRIGGER_TYPES + + def test_max_rationale_len_positive(self): + assert MAX_RATIONALE_LEN > 0 + + def test_max_notes_len_positive(self): + assert MAX_NOTES_LEN > 0 + + def test_update_directions_has_five_values(self): + assert len(VALID_UPDATE_DIRECTIONS) == 5 + + def test_trigger_types_has_seven_values(self): + assert len(VALID_CALIBRATION_TRIGGER_TYPES) == 7 + + +class TestPCUDataclass: + def test_valid_record_created(self): + record = _make_valid_record() + assert record.pcu_id == "PCU-0001" + + def test_dry_lab_only_false(self): + record = _make_valid_record() + assert record.dry_lab_only is False + + def test_human_reviewed_true(self): + record = _make_valid_record() + assert record.human_reviewed is True + + def test_n_whr_records_set(self): + record = _make_valid_record() + assert record.n_whr_records == 2 + + def test_whr_ids_is_list(self): + record = _make_valid_record() + assert isinstance(record.whr_ids, list) + + def test_limitations_is_list(self): + record = _make_valid_record() + assert isinstance(record.limitations, list) + + +class TestValidatePCU: + def test_valid_record_passes(self): + result = validate_post_experiment_calibration_update(_make_valid_record()) + assert result.valid is True + + def test_valid_record_no_violations(self): + result = validate_post_experiment_calibration_update(_make_valid_record()) + assert result.violations == [] + + def test_dry_lab_only_true_fails(self): + record = _make_valid_record(dry_lab_only=True) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_dry_lab_only_true_violation_message(self): + record = _make_valid_record(dry_lab_only=True) + result = validate_post_experiment_calibration_update(record) + assert any("dry_lab_only must be False" in v for v in result.violations) + + def test_human_reviewed_false_fails(self): + record = _make_valid_record(human_reviewed=False) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_human_reviewed_false_violation_message(self): + record = _make_valid_record(human_reviewed=False) + result = validate_post_experiment_calibration_update(record) + assert any("human_reviewed must be True" in v for v in result.violations) + + def test_bad_pcu_id_fails(self): + record = _make_valid_record(pcu_id="NOTAPCU") + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_n_whr_records_zero_fails(self): + record = _make_valid_record(n_whr_records=0, whr_ids=[]) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_n_whr_records_mismatch_fails(self): + record = _make_valid_record(n_whr_records=3, whr_ids=["WHR-0001"]) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_empty_whr_ids_fails(self): + record = _make_valid_record(n_whr_records=0, whr_ids=[]) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_empty_update_date_fails(self): + record = _make_valid_record(update_date="") + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_invalid_trigger_type_fails(self): + record = _make_valid_record(trigger_type="unknown_trigger") + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_invalid_update_direction_fails(self): + record = _make_valid_record(update_direction="random_direction") + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_update_magnitude_negative_fails(self): + record = _make_valid_record(update_magnitude=-0.1) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_update_magnitude_above_one_fails(self): + record = _make_valid_record(update_magnitude=1.1) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_update_magnitude_zero_passes(self): + record = _make_valid_record(update_magnitude=0.0, update_direction="no_change") + result = validate_post_experiment_calibration_update(record) + assert result.valid is True + + def test_update_magnitude_one_passes(self): + record = _make_valid_record(update_magnitude=1.0) + result = validate_post_experiment_calibration_update(record) + assert result.valid is True + + def test_update_magnitude_nan_fails(self): + record = _make_valid_record(update_magnitude=float("nan")) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_empty_rationale_fails(self): + record = _make_valid_record(rationale="") + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_rationale_too_long_fails(self): + record = _make_valid_record(rationale="x" * (MAX_RATIONALE_LEN + 1)) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_empty_limitations_fails(self): + record = _make_valid_record(limitations=[]) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_negative_prediction_error_before_fails(self): + record = _make_valid_record(prediction_error_before=-0.1) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_negative_prediction_error_after_fails(self): + record = _make_valid_record(prediction_error_after=-0.1) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_notes_too_long_fails(self): + record = _make_valid_record(notes="x" * (MAX_NOTES_LEN + 1)) + result = validate_post_experiment_calibration_update(record) + assert result.valid is False + + def test_all_valid_directions_pass(self): + for direction in VALID_UPDATE_DIRECTIONS: + record = _make_valid_record(update_direction=direction) + result = validate_post_experiment_calibration_update(record) + assert result.valid, f"Expected VALID for direction={direction}, got {result.violations}" + + def test_all_valid_trigger_types_pass(self): + for trigger in VALID_CALIBRATION_TRIGGER_TYPES: + record = _make_valid_record(trigger_type=trigger) + result = validate_post_experiment_calibration_update(record) + assert result.valid, f"Expected VALID for trigger_type={trigger}, got {result.violations}" + + def test_returns_pcu_validation_result(self): + result = validate_post_experiment_calibration_update(_make_valid_record()) + assert isinstance(result, PCUValidationResult) + + def test_pcu_id_in_result(self): + result = validate_post_experiment_calibration_update(_make_valid_record()) + assert result.pcu_id == "PCU-0001" + + +class TestBuildPCU: + def test_build_returns_record(self): + record = build_post_experiment_calibration_update( + pcu_id="PCU-0001", + update_date="2026-07-10", + whr_ids=["WHR-0001"], + trigger_type="hit_rate_above_baseline", + update_direction="expand_candidate_pool", + update_magnitude=0.1, + rationale="Hit rate exceeded baseline.", + limitations=["Single batch only."], + ) + assert isinstance(record, PostExperimentCalibrationUpdate) + + def test_build_enforces_dry_lab_only_false(self): + record = build_post_experiment_calibration_update( + pcu_id="PCU-0001", + update_date="2026-07-10", + whr_ids=["WHR-0001"], + trigger_type="hit_rate_above_baseline", + update_direction="no_change", + update_magnitude=0.0, + rationale="No change warranted.", + limitations=["Insufficient data."], + ) + assert record.dry_lab_only is False + + def test_build_enforces_human_reviewed_true(self): + record = build_post_experiment_calibration_update( + pcu_id="PCU-0002", + update_date="2026-07-10", + whr_ids=["WHR-0001", "WHR-0002"], + trigger_type="safety_signal_observed", + update_direction="restrict_candidate_pool", + update_magnitude=0.2, + rationale="Safety signal observed.", + limitations=["Limited n=2 observations."], + ) + assert record.human_reviewed is True + + def test_build_sets_n_whr_records_from_list(self): + record = build_post_experiment_calibration_update( + pcu_id="PCU-0003", + update_date="2026-07-10", + whr_ids=["WHR-0001", "WHR-0002", "WHR-0003"], + trigger_type="scheduled_review", + update_direction="no_change", + update_magnitude=0.0, + rationale="Scheduled review.", + limitations=["Insufficient data."], + ) + assert record.n_whr_records == 3 + + def test_build_validates_successfully(self): + record = build_post_experiment_calibration_update( + pcu_id="PCU-0004", + update_date="2026-07-10", + whr_ids=["WHR-0001"], + trigger_type="novel_family_detected", + update_direction="expand_candidate_pool", + update_magnitude=0.05, + rationale="Novel family detected in pilot batch.", + limitations=["Single batch observation."], + ) + result = validate_post_experiment_calibration_update(record) + assert result.valid is True + + +class TestFormatPCU: + def test_format_returns_string(self): + record = _make_valid_record() + output = format_post_experiment_calibration_update(record) + assert isinstance(output, str) + + def test_format_contains_pcu_id(self): + record = _make_valid_record() + output = format_post_experiment_calibration_update(record) + assert "PCU-0001" in output + + def test_format_contains_trigger_type(self): + record = _make_valid_record() + output = format_post_experiment_calibration_update(record) + assert "hit_rate_above_baseline" in output + + def test_format_contains_human_reviewed(self): + record = _make_valid_record() + output = format_post_experiment_calibration_update(record) + assert "human_reviewed" in output + + def test_format_valid_shows_valid(self): + record = _make_valid_record() + output = format_post_experiment_calibration_update(record) + assert "VALID" in output + + def test_format_invalid_shows_invalid(self): + record = _make_valid_record(human_reviewed=False) + output = format_post_experiment_calibration_update(record) + assert "INVALID" in output or "VIOLATION" in output diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index 369e70a4..f234d46c 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -3,7 +3,7 @@ import subprocess import math -BASELINE = 8711 +BASELINE = 8774 def test_test_count_regression():