From 16ec8c13474acaa479e125d8f735e70232377b05 Mon Sep 17 00:00:00 2001 From: OpenCode Date: Fri, 10 Jul 2026 12:05:24 +0700 Subject: [PATCH] feat: Phase I I7 batch comparative summary -- BCS- schema tracks trajectory across batches, trend/quality/calibration, dry-lab enforcement, 63 tests (#917) --- .../evidence/batch_comparative_summary.py | 212 ++++++++++ .../test_batch_comparative_summary.py | 397 ++++++++++++++++++ 2 files changed, 609 insertions(+) create mode 100644 src/openamp_foundry/evidence/batch_comparative_summary.py create mode 100644 tests/evidence/test_batch_comparative_summary.py diff --git a/src/openamp_foundry/evidence/batch_comparative_summary.py b/src/openamp_foundry/evidence/batch_comparative_summary.py new file mode 100644 index 00000000..1fb2d554 --- /dev/null +++ b/src/openamp_foundry/evidence/batch_comparative_summary.py @@ -0,0 +1,212 @@ +"""Comparative summary schema across multiple candidate batches.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +BATCH_COMPARATIVE_SUMMARY_ID_PREFIX = "BCS-" + +VALID_TREND_DIRECTIONS: frozenset = frozenset( + {"improving", "stable", "declining", "insufficient_data"} +) + +VALID_COMPARISON_STATUSES: frozenset = frozenset( + {"draft", "complete", "under_review", "finalized"} +) + +VALID_QUALITY_TIERS: frozenset = frozenset( + {"high", "medium", "low", "insufficient"} +) + +MIN_BATCHES_FOR_TREND: int = 2 +MAX_BATCHES_PER_SUMMARY: int = 20 + + +@dataclass +class BatchSnapshotEntry: + batch_id: str + batch_index: int + total_candidates: int + n_nominated: int + n_rejected: int + hit_rate: float + primary_rejection_reason: str + has_calibration_update: bool + snapshot_note: str = "" + + +@dataclass +class BatchComparativeSummary: + summary_id: str + batches: list = field(default_factory=list) + total_batches: int = 0 + overall_trend: str = "insufficient_data" + comparison_status: str = "draft" + dry_lab_only: bool = True + quality_tier: str = "insufficient" + trend_note: str = "" + improvement_observed: bool = False + is_example_data: bool = True + + +@dataclass +class ComparativeSummaryValidationResult: + is_valid: bool + violations: list = field(default_factory=list) + + +def build_batch_snapshot_entry( + batch_id: str, + batch_index: int, + total_candidates: int, + n_nominated: int, + n_rejected: int, + hit_rate: float, + primary_rejection_reason: str, + has_calibration_update: bool = False, + snapshot_note: str = "", +) -> BatchSnapshotEntry: + return BatchSnapshotEntry( + batch_id=batch_id, + batch_index=batch_index, + total_candidates=total_candidates, + n_nominated=n_nominated, + n_rejected=n_rejected, + hit_rate=hit_rate, + primary_rejection_reason=primary_rejection_reason, + has_calibration_update=has_calibration_update, + snapshot_note=snapshot_note, + ) + + +def build_batch_comparative_summary( + summary_id: str, + batches: list, + overall_trend: str, + comparison_status: str, + dry_lab_only: bool, + quality_tier: str, + trend_note: str = "", + improvement_observed: bool = False, + is_example_data: bool = True, +) -> BatchComparativeSummary: + return BatchComparativeSummary( + summary_id=summary_id, + batches=list(batches), + total_batches=len(batches), + overall_trend=overall_trend, + comparison_status=comparison_status, + dry_lab_only=dry_lab_only, + quality_tier=quality_tier, + trend_note=trend_note, + improvement_observed=improvement_observed, + is_example_data=is_example_data, + ) + + +def validate_batch_comparative_summary( + summary: BatchComparativeSummary, +) -> ComparativeSummaryValidationResult: + violations: list[str] = [] + + if not summary.summary_id.startswith(BATCH_COMPARATIVE_SUMMARY_ID_PREFIX): + violations.append( + f"summary_id must start with '{BATCH_COMPARATIVE_SUMMARY_ID_PREFIX}', " + f"got '{summary.summary_id}'" + ) + + if summary.overall_trend not in VALID_TREND_DIRECTIONS: + violations.append( + f"overall_trend '{summary.overall_trend}' not in VALID_TREND_DIRECTIONS" + ) + + if summary.comparison_status not in VALID_COMPARISON_STATUSES: + violations.append( + f"comparison_status '{summary.comparison_status}' not in VALID_COMPARISON_STATUSES" + ) + + if summary.quality_tier not in VALID_QUALITY_TIERS: + violations.append( + f"quality_tier '{summary.quality_tier}' not in VALID_QUALITY_TIERS" + ) + + if summary.total_batches != len(summary.batches): + violations.append( + f"total_batches={summary.total_batches} does not match len(batches)={len(summary.batches)}" + ) + + if summary.total_batches > MAX_BATCHES_PER_SUMMARY: + violations.append( + f"total_batches={summary.total_batches} exceeds MAX_BATCHES_PER_SUMMARY={MAX_BATCHES_PER_SUMMARY}" + ) + + if ( + summary.overall_trend in {"improving", "declining"} + and summary.total_batches < MIN_BATCHES_FOR_TREND + ): + violations.append( + f"overall_trend='{summary.overall_trend}' requires at least " + f"{MIN_BATCHES_FOR_TREND} batches, got {summary.total_batches}" + ) + + if not summary.dry_lab_only: + violations.append( + "dry_lab_only must be True; wet-lab comparative summaries require human review" + ) + + if summary.improvement_observed and summary.overall_trend not in {"improving", "stable"}: + violations.append( + "improvement_observed=True requires overall_trend in {'improving', 'stable'}" + ) + + for i, batch in enumerate(summary.batches): + prefix = f"batch[{i}] (id={batch.batch_id})" + + if not batch.batch_id: + violations.append(f"{prefix}: batch_id must not be empty") + + if batch.total_candidates < 0: + violations.append(f"{prefix}: total_candidates must be >= 0") + + if batch.n_nominated + batch.n_rejected > batch.total_candidates: + violations.append( + f"{prefix}: n_nominated + n_rejected " + f"({batch.n_nominated + batch.n_rejected}) " + f"> total_candidates ({batch.total_candidates})" + ) + + if not (0.0 <= batch.hit_rate <= 1.0): + violations.append( + f"{prefix}: hit_rate={batch.hit_rate} outside [0.0, 1.0]" + ) + + return ComparativeSummaryValidationResult( + is_valid=len(violations) == 0, + violations=violations, + ) + + +def format_batch_comparative_summary(summary: BatchComparativeSummary) -> str: + lines: list[str] = [] + lines.append(f"Batch Comparative Summary ({summary.summary_id})") + lines.append(f" Status: {summary.comparison_status}") + lines.append(f" Quality tier: {summary.quality_tier}") + lines.append(f" Overall trend: {summary.overall_trend}") + lines.append(f" Improvement observed: {summary.improvement_observed}") + lines.append(f" Total batches: {summary.total_batches}") + lines.append(f" Dry-lab only: {summary.dry_lab_only}") + if summary.trend_note: + lines.append(f" Note: {summary.trend_note}") + if summary.batches: + lines.append(" Batch snapshots:") + for b in summary.batches: + lines.append( + f" [{b.batch_index}] {b.batch_id}: " + f"total={b.total_candidates}, " + f"nominated={b.n_nominated}, " + f"rejected={b.n_rejected}, " + f"hit_rate={b.hit_rate:.2f}" + ) + if b.has_calibration_update: + lines.append(f" calibration update applied") + return "\n".join(lines) + "\n" diff --git a/tests/evidence/test_batch_comparative_summary.py b/tests/evidence/test_batch_comparative_summary.py new file mode 100644 index 00000000..65877688 --- /dev/null +++ b/tests/evidence/test_batch_comparative_summary.py @@ -0,0 +1,397 @@ +"""Tests for src/openamp_foundry/evidence/batch_comparative_summary.py (63 tests).""" + +import pytest +from openamp_foundry.evidence.batch_comparative_summary import ( + BATCH_COMPARATIVE_SUMMARY_ID_PREFIX, + VALID_TREND_DIRECTIONS, + VALID_COMPARISON_STATUSES, + VALID_QUALITY_TIERS, + MIN_BATCHES_FOR_TREND, + MAX_BATCHES_PER_SUMMARY, + BatchSnapshotEntry, + BatchComparativeSummary, + ComparativeSummaryValidationResult, + build_batch_snapshot_entry, + build_batch_comparative_summary, + validate_batch_comparative_summary, + format_batch_comparative_summary, +) + + +def _valid_snapshot(batch_id="TOY-BATCH-001", batch_index=0): + return build_batch_snapshot_entry( + batch_id=batch_id, + batch_index=batch_index, + total_candidates=20, + n_nominated=5, + n_rejected=10, + hit_rate=0.25, + primary_rejection_reason="toxicity_flag", + has_calibration_update=False, + snapshot_note="Toy batch snapshot", + ) + + +def _valid_summary(batches=None, overall_trend="stable"): + if batches is None: + batches = [_valid_snapshot("TOY-BATCH-001", 0), _valid_snapshot("TOY-BATCH-002", 1)] + return build_batch_comparative_summary( + summary_id="BCS-001", + batches=batches, + overall_trend=overall_trend, + comparison_status="draft", + dry_lab_only=True, + quality_tier="medium", + trend_note="Stable performance over two toy batches.", + improvement_observed=False, + is_example_data=True, + ) + + +class TestConstants: + def test_prefix(self): + assert BATCH_COMPARATIVE_SUMMARY_ID_PREFIX == "BCS-" + + def test_valid_trend_directions_is_frozenset(self): + assert isinstance(VALID_TREND_DIRECTIONS, frozenset) + + def test_improving_in_trends(self): + assert "improving" in VALID_TREND_DIRECTIONS + + def test_stable_in_trends(self): + assert "stable" in VALID_TREND_DIRECTIONS + + def test_declining_in_trends(self): + assert "declining" in VALID_TREND_DIRECTIONS + + def test_insufficient_data_in_trends(self): + assert "insufficient_data" in VALID_TREND_DIRECTIONS + + def test_valid_comparison_statuses_is_frozenset(self): + assert isinstance(VALID_COMPARISON_STATUSES, frozenset) + + def test_draft_in_statuses(self): + assert "draft" in VALID_COMPARISON_STATUSES + + def test_finalized_in_statuses(self): + assert "finalized" in VALID_COMPARISON_STATUSES + + def test_valid_quality_tiers_is_frozenset(self): + assert isinstance(VALID_QUALITY_TIERS, frozenset) + + def test_high_in_tiers(self): + assert "high" in VALID_QUALITY_TIERS + + def test_insufficient_in_tiers(self): + assert "insufficient" in VALID_QUALITY_TIERS + + def test_min_batches_for_trend(self): + assert MIN_BATCHES_FOR_TREND == 2 + + def test_max_batches_per_summary(self): + assert MAX_BATCHES_PER_SUMMARY == 20 + + +class TestBuildBatchSnapshotEntry: + def test_returns_batch_snapshot_entry(self): + entry = _valid_snapshot() + assert isinstance(entry, BatchSnapshotEntry) + + def test_batch_id_stored(self): + entry = _valid_snapshot("TOY-BATCH-X") + assert entry.batch_id == "TOY-BATCH-X" + + def test_batch_index_stored(self): + entry = _valid_snapshot(batch_index=3) + assert entry.batch_index == 3 + + def test_total_candidates_stored(self): + entry = _valid_snapshot() + assert entry.total_candidates == 20 + + def test_hit_rate_stored(self): + entry = _valid_snapshot() + assert entry.hit_rate == 0.25 + + def test_has_calibration_update_stored(self): + entry = build_batch_snapshot_entry( + batch_id="TOY-001", + batch_index=0, + total_candidates=10, + n_nominated=2, + n_rejected=5, + hit_rate=0.2, + primary_rejection_reason="novelty", + has_calibration_update=True, + ) + assert entry.has_calibration_update is True + + def test_snapshot_note_stored(self): + entry = _valid_snapshot() + assert entry.snapshot_note == "Toy batch snapshot" + + +class TestBuildBatchComparativeSummary: + def test_returns_batch_comparative_summary(self): + summary = _valid_summary() + assert isinstance(summary, BatchComparativeSummary) + + def test_summary_id_stored(self): + summary = _valid_summary() + assert summary.summary_id == "BCS-001" + + def test_total_batches_computed(self): + summary = _valid_summary() + assert summary.total_batches == 2 + + def test_overall_trend_stored(self): + summary = _valid_summary(overall_trend="stable") + assert summary.overall_trend == "stable" + + def test_comparison_status_stored(self): + summary = _valid_summary() + assert summary.comparison_status == "draft" + + def test_dry_lab_only_stored(self): + summary = _valid_summary() + assert summary.dry_lab_only is True + + def test_quality_tier_stored(self): + summary = _valid_summary() + assert summary.quality_tier == "medium" + + def test_is_example_data_stored(self): + summary = _valid_summary() + assert summary.is_example_data is True + + def test_improvement_observed_stored(self): + summary = _valid_summary() + assert summary.improvement_observed is False + + def test_single_batch_stored(self): + summary = build_batch_comparative_summary( + summary_id="BCS-001", + batches=[_valid_snapshot()], + overall_trend="insufficient_data", + comparison_status="draft", + dry_lab_only=True, + quality_tier="insufficient", + ) + assert summary.total_batches == 1 + + +class TestValidateBatchComparativeSummary: + def test_valid_summary_passes(self): + result = validate_batch_comparative_summary(_valid_summary()) + assert result.is_valid + + def test_returns_validation_result(self): + result = validate_batch_comparative_summary(_valid_summary()) + assert isinstance(result, ComparativeSummaryValidationResult) + + def test_no_violations_on_valid(self): + result = validate_batch_comparative_summary(_valid_summary()) + assert result.violations == [] + + def test_wrong_prefix_fails(self): + summary = _valid_summary() + summary.summary_id = "XYZ-001" + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("BCS-" in v for v in result.violations) + + def test_invalid_trend_fails(self): + summary = _valid_summary() + summary.overall_trend = "unknown" + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("overall_trend" in v for v in result.violations) + + def test_invalid_comparison_status_fails(self): + summary = _valid_summary() + summary.comparison_status = "bogus" + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + + def test_invalid_quality_tier_fails(self): + summary = _valid_summary() + summary.quality_tier = "excellent" + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + + def test_total_batches_mismatch_fails(self): + summary = _valid_summary() + summary.total_batches = 99 + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("total_batches" in v for v in result.violations) + + def test_too_many_batches_fails(self): + batches = [_valid_snapshot(f"TOY-{i:03d}", i) for i in range(MAX_BATCHES_PER_SUMMARY + 1)] + summary = build_batch_comparative_summary( + summary_id="BCS-001", + batches=batches, + overall_trend="stable", + comparison_status="draft", + dry_lab_only=True, + quality_tier="medium", + ) + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("MAX_BATCHES_PER_SUMMARY" in v for v in result.violations) + + def test_improving_trend_with_single_batch_fails(self): + summary = build_batch_comparative_summary( + summary_id="BCS-001", + batches=[_valid_snapshot()], + overall_trend="improving", + comparison_status="draft", + dry_lab_only=True, + quality_tier="medium", + ) + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("MIN_BATCHES_FOR_TREND" in v or "batches" in v for v in result.violations) + + def test_declining_trend_with_single_batch_fails(self): + summary = build_batch_comparative_summary( + summary_id="BCS-001", + batches=[_valid_snapshot()], + overall_trend="declining", + comparison_status="draft", + dry_lab_only=True, + quality_tier="medium", + ) + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + + def test_insufficient_data_with_single_batch_passes(self): + summary = build_batch_comparative_summary( + summary_id="BCS-001", + batches=[_valid_snapshot()], + overall_trend="insufficient_data", + comparison_status="draft", + dry_lab_only=True, + quality_tier="insufficient", + ) + result = validate_batch_comparative_summary(summary) + assert result.is_valid + + def test_dry_lab_false_fails(self): + summary = _valid_summary() + summary.dry_lab_only = False + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("dry_lab_only" in v for v in result.violations) + + def test_improvement_observed_with_declining_fails(self): + summary = _valid_summary(overall_trend="declining") + summary.improvement_observed = True + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("improvement_observed" in v for v in result.violations) + + def test_improvement_observed_with_improving_passes(self): + summary = _valid_summary(overall_trend="improving") + summary.improvement_observed = True + result = validate_batch_comparative_summary(summary) + assert result.is_valid + + def test_batch_empty_id_fails(self): + batch = _valid_snapshot() + batch.batch_id = "" + summary = _valid_summary(batches=[batch, _valid_snapshot("TOY-002", 1)]) + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("batch_id" in v for v in result.violations) + + def test_batch_hit_rate_out_of_range_fails(self): + batch = _valid_snapshot() + batch.hit_rate = 1.5 + summary = _valid_summary(batches=[batch, _valid_snapshot("TOY-002", 1)]) + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("hit_rate" in v for v in result.violations) + + def test_batch_nominations_plus_rejections_exceed_total_fails(self): + batch = _valid_snapshot() + batch.n_nominated = 15 + batch.n_rejected = 15 + batch.total_candidates = 20 + summary = _valid_summary(batches=[batch, _valid_snapshot("TOY-002", 1)]) + result = validate_batch_comparative_summary(summary) + assert not result.is_valid + assert any("total_candidates" in v for v in result.violations) + + def test_multiple_violations_accumulated(self): + summary = _valid_summary() + summary.summary_id = "BAD-001" + summary.overall_trend = "unknown" + result = validate_batch_comparative_summary(summary) + assert len(result.violations) >= 2 + + def test_all_valid_trends_accepted(self): + for trend in VALID_TREND_DIRECTIONS: + batches = [_valid_snapshot("TOY-001", 0), _valid_snapshot("TOY-002", 1)] + summary = build_batch_comparative_summary( + summary_id="BCS-001", + batches=batches, + overall_trend=trend, + comparison_status="draft", + dry_lab_only=True, + quality_tier="medium", + ) + result = validate_batch_comparative_summary(summary) + assert result.is_valid, f"Trend {trend} failed: {result.violations}" + + +class TestFormatBatchComparativeSummary: + def test_returns_string(self): + summary = _valid_summary() + result = format_batch_comparative_summary(summary) + assert isinstance(result, str) + + def test_contains_summary_id(self): + summary = _valid_summary() + result = format_batch_comparative_summary(summary) + assert "BCS-001" in result + + def test_contains_trend(self): + summary = _valid_summary() + result = format_batch_comparative_summary(summary) + assert "stable" in result + + def test_contains_total_batches(self): + summary = _valid_summary() + result = format_batch_comparative_summary(summary) + assert "2" in result + + def test_contains_batch_ids(self): + summary = _valid_summary() + result = format_batch_comparative_summary(summary) + assert "TOY-BATCH-001" in result + assert "TOY-BATCH-002" in result + + def test_contains_hit_rate(self): + summary = _valid_summary() + result = format_batch_comparative_summary(summary) + assert "0.25" in result + + def test_shows_calibration_update(self): + batch = _valid_snapshot() + batch.has_calibration_update = True + summary = build_batch_comparative_summary( + summary_id="BCS-001", + batches=[batch, _valid_snapshot("TOY-002", 1)], + overall_trend="stable", + comparison_status="draft", + dry_lab_only=True, + quality_tier="medium", + ) + result = format_batch_comparative_summary(summary) + assert "calibration" in result.lower() + + def test_ends_with_newline(self): + summary = _valid_summary() + result = format_batch_comparative_summary(summary) + assert result.endswith("\n")