From 3476a97d8bd2456fa7d6e4c0e6eff3a68e276058 Mon Sep 17 00:00:00 2001 From: OpenCode Date: Mon, 6 Jul 2026 10:25:05 +0700 Subject: [PATCH] =?UTF-8?q?v0.5.45=20=E2=80=94=20Active-learning=20batch-2?= =?UTF-8?q?=20selector=20(Loop=2024)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add batch-2 candidate selector for the calibration loop: - active_learning/selector.py: select_batch_2() with uncertainty sampling, diversity vs batch-1, configurable safety/selectivity gates, and min-uncertainty-probe guarantee - CLI : reads pool CSV and batch-1 IDs, writes JSON manifest - 11 tests covering: basic selection, safety gate, selectivity gate, uncertainty probe, diversity, empty pool, all-gated-out, selection_reason, metadata shape, CLI success/error paths - 1824 total tests passing --- docs/50_LOOP_PLAN.md | 10 +- docs/ROADMAP.md | 24 ++ .../active_learning/__init__.py | 22 ++ .../active_learning/selector.py | 312 ++++++++++++++++++ src/openamp_foundry/cli/commands/selection.py | 53 +++ src/openamp_foundry/cli/main.py | 73 +++- tests/test_active_learning.py | 286 ++++++++++++++++ 7 files changed, 774 insertions(+), 6 deletions(-) create mode 100644 src/openamp_foundry/active_learning/__init__.py create mode 100644 src/openamp_foundry/active_learning/selector.py create mode 100644 tests/test_active_learning.py diff --git a/docs/50_LOOP_PLAN.md b/docs/50_LOOP_PLAN.md index 54315e1b..4c4435f6 100644 --- a/docs/50_LOOP_PLAN.md +++ b/docs/50_LOOP_PLAN.md @@ -1,8 +1,8 @@ # 50-Loop Execution Plan -> **Status:** Strategic roadmap. Updated v0.5.44. -> **Current state:** 1813 tests, pipeline AUROC 0.7792, calibration intake + gate + engine + dry-run + policy-version + synthetic generator + recalibration report shipped, -> Phase 0 complete (Loops 1–8), Phase 1 complete (Loops 9–17), Phase 2 Loops 18, 19, 20, 21, 22, 23 complete, +> **Status:** Strategic roadmap. Updated v0.5.45. +> **Current state:** 1824 tests, pipeline AUROC 0.7792, calibration intake + gate + engine + dry-run + policy-version + synthetic generator + recalibration report + batch-2 selector shipped, +> Phase 0 complete (Loops 1–8), Phase 1 complete (Loops 9–17), Phase 2 Loops 18, 19, 20, 21, 22, 23, 24 complete, > cluster-split/selectivity/triage now gated in CI, Wave 0.5 panel ready (24 candidates, 15 families), no wet-lab data yet. > > Each loop = one focused PR: bottleneck identified → implemented → verified → merged. @@ -74,7 +74,7 @@ Build the recalibration engine gated by the policy. Implement active-learning ba | 21 ✅ | **Recalibration engine**: the gate existed but no engine. Real weight-update code needs to be safe, auditable, and gated | `calibration/engine.py`: `compute_weight_update(intake_report, gate_verdict, current_weights, l1_budget)` returns `WeightUpdateProposal`. Control theory: delta = learning_rate × (observed_accuracy − target_accuracy). Conservative learning rate 0.05. L1 budget enforced. v0.5.36 | 1735 tests; CLI `recalibration-engine` exits 0 on proposal, 3 on violations, 2 on missing files; 12 engine tests | | 22 ✅ | No "dry run" mode for recalibration engine. User can't preview what weights would change | `--dry-run` flag: compute and log proposed weight changes without applying them. Outputs comparison table (current vs proposed). v0.5.43. 1 test, Makefile target. | Dry run produces diff without side effects; verified with `make recalibration-engine-dry-run` + manual inspection | | 23 ✅ | No weight-change report that a human reviewer can inspect | `reports/recalibration_report.py`: generates human-readable report with before/after weights, rationale, policy-rule results, gate-verdict context, and explicit "human review required" banner. Schema validation via `schemas/recalibration_report.schema.json`. v0.5.44. 9 tests. | Report validates against schema; gate verdict details preserved; CLI uses combined report for JSON/MD output | -| 24 | No second-batch selection logic. The gate allows recalibration, but how do we pick the next 8–12 candidates? | `active_learning/selector.py`: implements uncertainty sampling + diversity + safety gate. Returns ranked candidates for batch 2 | Selection respects safety constraints; top-5 include at least 1 high-uncertainty probe | +| 24 ✅ | No second-batch selection logic. The gate allows recalibration, but how do we pick the next 8–12 candidates? | `active_learning/selector.py`: implements uncertainty sampling + diversity + safety gate. CLI `select-batch`. v0.5.45. 11 tests. | Selection respects safety constraints; top-5 include at least 1 high-uncertainty probe; all blocked candidates rejected; CLI produces valid JSON manifest | | 25 | No active-learning benchmark on synthetic data. Can the selector recover known "hidden active" candidates better than random? | Synthetic benchmark: hide 3 known active AMPs from the panel, run selection, measure how many rounds to recover them | Recovery within N rounds (N pre-registered) | | 26 | No integration between active-learning selector and the calibration pipeline | End-to-end flow: intake → gate (pass) → recalibration engine (dry-run) → selector → batch 2 manifest | Full chain runs on synthetic data, produces a batch-2 manifest | | 27 | No end-to-end regression test for the full calibration loop (the "golden path") | `tests/test_calibration_full_loop.py`: generate synthetic results → intake → gate → dry-run recalibration → selector → batch manifest. Assert all exit codes 0 | Full golden path tested on every PR | @@ -177,7 +177,7 @@ If no data arrives, virtual assay scaffolding continues independently. ``` Phase 0: ✅ Complete (Loops 1–8) Phase 1: ✅ Complete (Loops 9–17) -Phase 2: Loop 23 of 29 (recalibration report shipped — Loop 24 next) +Phase 2: Loop 24 of 29 (batch-2 selector shipped — Loop 25 next) Phase 3: Not started (Loops 30–39) Phase 4: Not started (Loops 40–49) ``` diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 95541cf2..a9f8e2e7 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -1,5 +1,29 @@ # Roadmap +## v0.5.45 — Active-Learning Batch-2 Selector ✓ (2026-07-06) + +The recalibration pipeline could compute weight updates and produce a reviewable +report, but had no way to select the next set of candidates for lab testing. This +release adds a standalone batch-2 selector that uses uncertainty sampling, +diversity ranking, and a safety gate to pick informative candidates. + +Changes: +- Added ``active_learning/selector.py`` — ``select_batch_2()`` combines + per-candidate ensemble score, uncertainty (from model disagreement and + ensemble proximity to 0.5), and diversity vs batch 1 (via normalized sequence + similarity). The safety gate rejects candidates below configurable safety and + selectivity thresholds. At least 1 high-uncertainty probe is guaranteed in + the top-N selection. +- Added ``active_learning/__init__.py`` — exports ``BatchSelection`` and + ``select_batch_2``. +- Added CLI ``select-batch`` — reads a candidate pool CSV and batch-1 IDs, + writes a JSON manifest with selected candidates and metadata (version, + probes_in_top_n, n_after_safety_gate, notes). +- Added 11 tests: basic selection, safety gate, selectivity gate, uncertainty + probe guarantee, diversity vs batch-1, empty pool, all-gated-out, + selection_reason field, metadata shape, CLI success, CLI error path. +- Updated CLI parser (``main.py``) and handler (``selection.py``). + ## v0.5.44 — Recalibration Report with Schema Validation ✓ (2026-07-06) The existing weight-update proposal markdown writer was embedded in engine.py diff --git a/src/openamp_foundry/active_learning/__init__.py b/src/openamp_foundry/active_learning/__init__.py new file mode 100644 index 00000000..c439e9e2 --- /dev/null +++ b/src/openamp_foundry/active_learning/__init__.py @@ -0,0 +1,22 @@ +"""Active learning: second-batch candidate selection for assay panels. + +After an initial batch is tested in the lab and the recalibration pipeline +computes weight updates, the active-learning selector picks the next batch +of candidates. It balances: + +- **Uncertainty**: candidates where the model is most unsure (high + disagreement, or ensemble score near 0.5). +- **Diversity**: candidates that are structurally distinct from the first + batch, to explore new sequence space. +- **Safety**: candidates that pass a minimum safety/toxicity threshold. +""" + +from openamp_foundry.active_learning.selector import ( + BatchSelection, + select_batch_2, +) + +__all__ = [ + "BatchSelection", + "select_batch_2", +] diff --git a/src/openamp_foundry/active_learning/selector.py b/src/openamp_foundry/active_learning/selector.py new file mode 100644 index 00000000..af603232 --- /dev/null +++ b/src/openamp_foundry/active_learning/selector.py @@ -0,0 +1,312 @@ +"""Batch-2 candidate selector: uncertainty, diversity, safety gating. + +Picks the next N candidates for lab testing after an initial batch has been +assayed and the recalibration pipeline has run. This is a **selection triage** +module, not a scorer or predictor. All scores are pipeline-produced. +""" + +from __future__ import annotations + +import csv +import logging +from dataclasses import dataclass, field +from pathlib import Path + +_log = logging.getLogger(__name__) + +_SELECTOR_VERSION = "0.1.0" + + +@dataclass(frozen=True) +class BatchSelection: + """Result of batch-2 selection.""" + + version: str + n_requested: int + n_remaining: int + n_after_safety_gate: int + selected: tuple[dict, ...] + probes_in_top_n: int + notes: tuple[str, ...] + + def to_dict(self) -> dict: + return { + "version": self.version, + "n_requested": self.n_requested, + "n_remaining": self.n_remaining, + "n_after_safety_gate": self.n_after_safety_gate, + "selected": list(self.selected), + "probes_in_top_n": self.probes_in_top_n, + "notes": list(self.notes), + } + + +def _load_candidates(csv_path: Path) -> list[dict]: + """Load candidates from a panel CSV.""" + rows: list[dict] = [] + with open(csv_path, newline="") as f: + reader = csv.DictReader(f) + for row in reader: + parsed: dict[str, float | int | str] = {} + for k, v in row.items(): + k = k.strip() + if v is None: + parsed[k] = None + else: + v = v.strip() + # Try numeric conversion + try: + parsed[k] = float(v) + except (ValueError, TypeError): + try: + parsed[k] = int(v) + except (ValueError, TypeError): + parsed[k] = v + rows.append(parsed) + return rows + + +def _safe_float(val: object, default: float = 0.0) -> float: + if isinstance(val, (int, float)): + return float(val) + return default + + +def _compute_uncertainty(row: dict) -> float: + """Compute an uncertainty score in [0, 1] for a candidate. + + Uses the pipeline's built-in ``disagreement`` field (when available) + and complements it with the distance of the ``ensemble`` score from 0.5 + (the most uncertain region for a binary classifier). + + A value of 1.0 means maximum uncertainty; 0.0 means complete certainty. + """ + disagreement = _safe_float(row.get("disagreement"), default=0.0) + + ensemble = _safe_float(row.get("ensemble"), default=0.5) + ensemble_distance = 1.0 - 2.0 * abs(ensemble - 0.5) + ensemble_distance = max(0.0, min(1.0, ensemble_distance)) + + return max(disagreement, ensemble_distance) + + +def _compute_diversity_vs_batch( + sequence: str, + batch_1_sequences: list[str], +) -> float: + """Compute diversity score for a candidate relative to a set of sequences. + + Uses ``normalized_similarity`` from the novelty module. A score of 1.0 + means the candidate is completely novel relative to batch 1 (maximally + diverse); 0.0 means it is identical to at least one batch-1 candidate. + """ + if not batch_1_sequences or not sequence: + return 1.0 + from openamp_foundry.scoring.novelty import normalized_similarity + + max_sim = max( + normalized_similarity(sequence, other) + for other in batch_1_sequences + ) + return 1.0 - max_sim + + +def _passes_safety_gate( + row: dict, + safety_threshold: float = 0.5, + selectivity_threshold: float = 0.5, +) -> tuple[bool, str]: + """Check whether a candidate passes the safety gate. + + Returns (passes, reason). + """ + reasons: list[str] = [] + + safety = _safe_float(row.get("safety"), default=1.0) + if safety < safety_threshold: + reasons.append( + f"safety={safety:.3f} < {safety_threshold:.3f}" + ) + + selectivity = _safe_float(row.get("rich_selectivity")) + if not selectivity: + selectivity = _safe_float(row.get("selectivity_proxy"), default=1.0) + if selectivity < selectivity_threshold: + reasons.append( + f"selectivity={selectivity:.3f} < {selectivity_threshold:.3f}" + ) + + if reasons: + return False, "; ".join(reasons) + return True, "passes gate" + + +def select_batch_2( + candidates_csv: str | Path, + batch_1_ids: list[str], + *, + n: int = 10, + safety_threshold: float = 0.5, + selectivity_threshold: float = 0.5, + ensemble_weight: float = 0.40, + uncertainty_weight: float = 0.30, + diversity_weight: float = 0.30, + min_uncertainty_probes: int = 1, +) -> BatchSelection: + """Select the next batch of candidates for lab testing. + + Args: + candidates_csv: Path to the full candidate pool CSV (same format as + the pilot panel CSV with columns: candidate_id, sequence, + ensemble, activity, disagreement, safety, novelty, etc.). + batch_1_ids: List of candidate IDs already tested in batch 1. + n: Desired batch-2 size. + safety_threshold: Minimum ``safety`` score (0-1) for a candidate to + be eligible. Candidates below this are blocked. + selectivity_threshold: Minimum ``rich_selectivity`` (or + ``selectivity_proxy`` as fallback) score. + ensemble_weight: Weight for the ensemble score in the combined rank. + uncertainty_weight: Weight for the uncertainty score. + diversity_weight: Weight for the diversity vs batch-1 score. + min_uncertainty_probes: At least this many high-uncertainty + candidates must appear in the final selection (to actively probe + model blind spots). + + Returns: + ``BatchSelection`` with the selected candidates (enriched with + uncertainty, diversity, and selection reason metadata). + """ + csv_path = Path(candidates_csv) + if not csv_path.exists(): + raise FileNotFoundError(f"Candidates CSV not found: {csv_path}") + + all_candidates = _load_candidates(csv_path) + + batch_1_set = set(batch_1_ids) + batch_1_rows = [c for c in all_candidates if c.get("candidate_id") in batch_1_set] + batch_1_sequences = [str(c.get("sequence", "")) for c in batch_1_rows] + + remaining = [ + c for c in all_candidates + if c.get("candidate_id") not in batch_1_set + ] + + if not remaining: + return BatchSelection( + version=_SELECTOR_VERSION, + n_requested=n, + n_remaining=0, + n_after_safety_gate=0, + selected=(), + probes_in_top_n=0, + notes=("No remaining candidates to select from.",), + ) + + # Compute scores and gate + scored: list[tuple[float, dict, float, float, float, str]] = [] + n_gated_out = 0 + for row in remaining: + passes, reason = _passes_safety_gate( + row, + safety_threshold=safety_threshold, + selectivity_threshold=selectivity_threshold, + ) + if not passes: + n_gated_out += 1 + _log.debug( + "Candidate %s gated out: %s", + row.get("candidate_id", "?"), reason, + ) + continue + + ensemble_score = _safe_float(row.get("ensemble"), default=0.0) + uncertainty = _compute_uncertainty(row) + diversity = _compute_diversity_vs_batch( + str(row.get("sequence", "")), + batch_1_sequences, + ) + + combined = ( + ensemble_weight * ensemble_score + + uncertainty_weight * uncertainty + + diversity_weight * diversity + ) + + scored.append((combined, row, ensemble_score, uncertainty, diversity, reason)) + + if not scored: + return BatchSelection( + version=_SELECTOR_VERSION, + n_requested=n, + n_remaining=len(remaining), + n_after_safety_gate=0, + selected=(), + probes_in_top_n=0, + notes=(f"All {len(remaining)} remaining candidates failed the safety gate.",), + ) + + # Sort by combined score descending + scored.sort(key=lambda x: x[0], reverse=True) + + # Build selection with uncertainty probe guarantee + selected_list: list[dict] = [] + notes: list[str] = [] + + all_uncertainty_ranked = sorted( + [s for s in scored if s[3] >= 0.6], + key=lambda x: x[3], reverse=True, + ) + + probes_placed = 0 + taken_ids: set[str] = set() + + # Phase 1: place high-uncertainty probes if needed + for _, row, _, _, _, _reason in all_uncertainty_ranked: + if probes_placed >= min_uncertainty_probes: + break + cid = str(row.get("candidate_id", "")) + if cid not in taken_ids: + taken_ids.add(cid) + row_meta = dict(row) + row_meta["selection_reason"] = "high_uncertainty_probe" + selected_list.append(row_meta) + probes_placed += 1 + + # Phase 2: fill remaining slots from top-ranked pool (skip taken) + for _, row, ens, unc, div, _reason in scored: + if len(selected_list) >= n: + break + cid = str(row.get("candidate_id", "")) + if cid in taken_ids: + continue + taken_ids.add(cid) + row_meta = dict(row) + row_meta["selection_reason"] = ( + f"combined={ens * ensemble_weight + unc * uncertainty_weight + div * diversity_weight:.4f}; " + f"ensemble={ens:.3f}, uncertainty={unc:.3f}, diversity={div:.3f}" + ) + selected_list.append(row_meta) + + if probes_placed < min_uncertainty_probes: + notes.append( + f"Only {probes_placed}/{min_uncertainty_probes} high-uncertainty probes " + f"found in eligible pool; fewer probes placed than requested." + ) + + if len(selected_list) < n: + notes.append( + f"Requested {n} but only {len(selected_list)} candidates " + f"available after safety gating and deduplication." + ) + + n_after_gate = len(scored) + + return BatchSelection( + version=_SELECTOR_VERSION, + n_requested=n, + n_remaining=len(remaining), + n_after_safety_gate=n_after_gate, + selected=tuple(selected_list), + probes_in_top_n=probes_placed, + notes=tuple(notes), + ) diff --git a/src/openamp_foundry/cli/commands/selection.py b/src/openamp_foundry/cli/commands/selection.py index f19c76c0..c87042d9 100644 --- a/src/openamp_foundry/cli/commands/selection.py +++ b/src/openamp_foundry/cli/commands/selection.py @@ -344,3 +344,56 @@ def _run_diversity_check(args: argparse.Namespace) -> int: }, indent=2)) return 0 + +def _run_select_batch(args: argparse.Namespace) -> int: + """Select a second batch of candidates for lab testing. + + Reads the full candidate pool CSV, filters out batch-1 candidates, + applies safety gating, uncertainty sampling, and diversity ranking. + Writes a JSON manifest. + """ + from openamp_foundry.active_learning.selector import select_batch_2 + + pool_path = Path(args.candidates) + if not pool_path.exists(): + print(json.dumps({"status": "error", "error": f"Candidates CSV not found: {pool_path}"})) + return 1 + + batch_1_ids = [s.strip() for s in args.batch_1_ids.split(",") if s.strip()] + if not batch_1_ids: + print(json.dumps({"status": "error", "error": "batch-1-ids must be a non-empty comma-separated list"})) + return 1 + + result = select_batch_2( + candidates_csv=pool_path, + batch_1_ids=batch_1_ids, + n=args.n, + safety_threshold=args.safety_threshold, + selectivity_threshold=args.selectivity_threshold, + ensemble_weight=args.ensemble_weight, + uncertainty_weight=args.uncertainty_weight, + diversity_weight=args.diversity_weight, + min_uncertainty_probes=args.min_uncertainty_probes, + ) + + out_path = Path(args.out) + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_text( + json.dumps(result.to_dict(), indent=2, sort_keys=False) + "\n" + ) + + selected_ids = [c.get("candidate_id", "?") for c in result.selected] + + print(json.dumps({ + "status": "ok", + "n_requested": args.n, + "n_remaining": result.n_remaining, + "n_after_safety_gate": result.n_after_safety_gate, + "n_selected": len(result.selected), + "probes_in_top_n": result.probes_in_top_n, + "selected_ids": selected_ids, + "notes": list(result.notes), + "out": str(out_path), + }, indent=2)) + return 0 + diff --git a/src/openamp_foundry/cli/main.py b/src/openamp_foundry/cli/main.py index 7a98c411..9dd5af07 100644 --- a/src/openamp_foundry/cli/main.py +++ b/src/openamp_foundry/cli/main.py @@ -2,7 +2,7 @@ from openamp_foundry.cli.commands.core import _run_generate_batch from openamp_foundry.cli.commands.benchmark import _run_bench, _run_validate_scoring, _run_cluster_split_bench, _run_expert_ablation_bench, _run_selectivity_bench, _run_triage, _run_metrics_snapshot, _run_feature_decomp -from openamp_foundry.cli.commands.selection import _run_pilot_panel, _run_pilot_confident, _run_diversity_check +from openamp_foundry.cli.commands.selection import _run_pilot_panel, _run_pilot_confident, _run_diversity_check, _run_select_batch from openamp_foundry.cli.commands.external import _run_external_predict, _run_external_consensus from openamp_foundry.cli.commands.qc import _run_synthesis_order, _run_presynth_qc from openamp_foundry.cli.commands.reports import _run_reviewer_questionnaire, _run_ip_report, _run_batch_pack, _run_gold_standard, _run_novelty_check_broad, _run_lab_result_report, _run_calibration_intake, _run_recalibration_gate, _run_recalibration_engine, _run_validate_policy_version @@ -884,6 +884,74 @@ def build_parser() -> argparse.ArgumentParser: help="ISO date (YYYY-MM-DD) for 'today'. Defaults to actual today.", ) + # ── Batch-2 selector ────────────────────────────────────────────── + select_batch = sub.add_parser( + "select-batch", + help=( + "Select the next batch of candidates for lab testing after " + "an initial batch has been assayed and the recalibration " + "pipeline has run. Uses uncertainty sampling + diversity + " + "safety gating to pick informative candidates." + ), + ) + select_batch.add_argument( + "--candidates", + required=True, + help="Path to the full candidate pool CSV (same format as pilot panel).", + ) + select_batch.add_argument( + "--batch-1-ids", + required=True, + help="Comma-separated candidate IDs already tested in batch 1.", + ) + select_batch.add_argument( + "--n", + type=int, + default=10, + help="Desired batch-2 size (default: 10).", + ) + select_batch.add_argument( + "--safety-threshold", + type=float, + default=0.5, + help="Minimum safety score (0-1) for eligibility (default: 0.5).", + ) + select_batch.add_argument( + "--selectivity-threshold", + type=float, + default=0.5, + help="Minimum rich_selectivity score for eligibility (default: 0.5).", + ) + select_batch.add_argument( + "--ensemble-weight", + type=float, + default=0.40, + help="Weight for ensemble score in combined rank (default: 0.40).", + ) + select_batch.add_argument( + "--uncertainty-weight", + type=float, + default=0.30, + help="Weight for uncertainty score (default: 0.30).", + ) + select_batch.add_argument( + "--diversity-weight", + type=float, + default=0.30, + help="Weight for diversity vs batch-1 (default: 0.30).", + ) + select_batch.add_argument( + "--min-uncertainty-probes", + type=int, + default=1, + help="At least this many high-uncertainty candidates in the selection (default: 1).", + ) + select_batch.add_argument( + "--out", + required=True, + help="Output path for the batch-2 manifest JSON.", + ) + return parser @@ -999,6 +1067,9 @@ def main(argv: list[str] | None = None) -> int: if args.command == "validate-policy-version": return _run_validate_policy_version(args) + if args.command == "select-batch": + return _run_select_batch(args) + parser.error("unknown command") return 2 diff --git a/tests/test_active_learning.py b/tests/test_active_learning.py new file mode 100644 index 00000000..5ff4502b --- /dev/null +++ b/tests/test_active_learning.py @@ -0,0 +1,286 @@ +"""Tests for the active-learning batch-2 selector. + +Honest limitation: All data is synthetic. These tests verify code-path +integrity, not biological sampling efficiency. +""" + +from __future__ import annotations + +import csv +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +# ── Helpers ─────────────────────────────────────────────────────────── + + +def _write_panel_csv(path: Path, rows: list[dict]) -> Path: + fieldnames = [ + "pilot_rank", "candidate_id", "sequence", "length", "seed", + "ensemble", "activity", "boman_activity", "disagreement", + "safety", "synthesis", "novelty", "serum_stability", + "selectivity_proxy", "rich_selectivity", "pilot_priority", + "amphipathic_score", "charge_ph74", + ] + with path.open("w", newline="") as f: + w = csv.DictWriter(f, fieldnames=fieldnames) + w.writeheader() + for r in rows: + w.writerow(r) + return path + + +def _panel_row( + candidate_id: str, + sequence: str, + seed: str = "SEED-TEST", + ensemble: float = 0.80, + disagreement: float = 0.10, + safety: float = 0.90, + rich_selectivity: float = 0.75, +) -> dict: + return { + "pilot_rank": "1", + "candidate_id": candidate_id, + "sequence": sequence, + "length": str(len(sequence)), + "seed": seed, + "ensemble": f"{ensemble:.2f}", + "activity": f"{ensemble:.2f}", + "boman_activity": "0.75", + "disagreement": f"{disagreement:.2f}", + "safety": f"{safety:.2f}", + "synthesis": "0.85", + "novelty": "0.70", + "serum_stability": "0.65", + "selectivity_proxy": "0.60", + "rich_selectivity": f"{rich_selectivity:.2f}", + "pilot_priority": "0.75", + "amphipathic_score": "1.5", + "charge_ph74": "4.0", + } + + +# ── Tests ───────────────────────────────────────────────────────────── + + +class TestSelectBatch2: + """Tests for ``select_batch_2()``.""" + + def test_basic_selection(self, tmp_path): + """Selects the correct number of candidates from a pool.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + rows = [ + _panel_row(f"CAND-{i:03d}", f"AAAAKKKKFF{i}", seed=f"SEED-{i}") + for i in range(20) + ] + _write_panel_csv(pool, rows) + batch_1 = ["CAND-000", "CAND-001", "CAND-002"] + + result = select_batch_2(pool, batch_1, n=5) + assert len(result.selected) == 5 + assert result.n_requested == 5 + assert result.n_remaining == 17 + assert all( + c.get("candidate_id") not in batch_1 + for c in result.selected + ) + + def test_safety_gate_filters_out_toxic(self, tmp_path): + """Candidates below safety threshold are excluded.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + rows = [ + _panel_row("CAND-SAFE-1", "AAAAKKKFF1", safety=0.9), + _panel_row("CAND-TOXIC-1", "BBBBLLLGG2", safety=0.2), + _panel_row("CAND-SAFE-2", "CCCCMMMHH3", safety=0.8), + _panel_row("CAND-TOXIC-2", "DDDDNNNII4", safety=0.1), + ] + _write_panel_csv(pool, rows) + + result = select_batch_2(pool, [], n=4) + ids = [c.get("candidate_id") for c in result.selected] + assert "CAND-TOXIC-1" not in ids + assert "CAND-TOXIC-2" not in ids + assert "CAND-SAFE-1" in ids + assert "CAND-SAFE-2" in ids + assert result.n_after_safety_gate == 2 + + def test_selectivity_gate_filters(self, tmp_path): + """Candidates below selectivity threshold are excluded.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + rows = [ + _panel_row("CAND-SEL-1", "AAAAKKKFF1", rich_selectivity=0.8), + _panel_row("CAND-LOW-SEL", "BBBBLLLGG2", rich_selectivity=0.3), + ] + _write_panel_csv(pool, rows) + + result = select_batch_2(pool, [], n=2) + ids = [c.get("candidate_id") for c in result.selected] + assert "CAND-LOW-SEL" not in ids + assert "CAND-SEL-1" in ids + + def test_uncertainty_probe_guarantee(self, tmp_path): + """At least one high-uncertainty candidate is selected when available.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + rows = [ + _panel_row("CAND-HIGH-UNC", "AAAAKKKFF1", ensemble=0.51, disagreement=0.45), + _panel_row("CAND-LOW-UNC-A", "BBBBLLLGG2", ensemble=0.90, disagreement=0.05), + _panel_row("CAND-LOW-UNC-B", "CCCCMMMHH3", ensemble=0.92, disagreement=0.04), + _panel_row("CAND-LOW-UNC-C", "DDDDNNNII4", ensemble=0.88, disagreement=0.06), + _panel_row("CAND-LOW-UNC-D", "EEEEOOOJJ5", ensemble=0.91, disagreement=0.03), + ] + _write_panel_csv(pool, rows) + + result = select_batch_2(pool, [], n=3, min_uncertainty_probes=1) + selected_ids = [c.get("candidate_id") for c in result.selected] + assert "CAND-HIGH-UNC" in selected_ids + assert result.probes_in_top_n >= 1 + + def test_diversity_vs_batch_1(self, tmp_path): + """Diverse candidates are preferred over near-duplicates of batch 1.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + rows = [ + _panel_row("CAND-NOVEL", "QWERTYUIOPLKJHGFDSA", ensemble=0.75, disagreement=0.10), + _panel_row("CAND-DUP", "AAAKKKFFFI", ensemble=0.85, disagreement=0.05), + ] + batch_1_candidates = tmp_path / "batch1.csv" + _write_panel_csv(pool, rows) + _write_panel_csv(batch_1_candidates, [ + _panel_row("BATCH1-1", "AAAKKKFFFI", ensemble=0.82, disagreement=0.07), + ]) + + result = select_batch_2( + pool, + batch_1_ids=["BATCH1-1"], + n=2, + diversity_weight=0.5, + uncertainty_weight=0.0, + ensemble_weight=0.5, + ) + selected_ids = [c.get("candidate_id") for c in result.selected] + # CAND-DUP is a near-duplicate of batch1-1, so with high diversity weight + # the novel candidate should rank higher + assert "CAND-NOVEL" in selected_ids + + def test_no_remaining_candidates(self, tmp_path): + """Empty pool returns empty selection with explanatory note.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + _write_panel_csv(pool, [ + _panel_row("CAND-001", "AAAAKKKFFF"), + ]) + + result = select_batch_2(pool, batch_1_ids=["CAND-001"], n=5) + assert len(result.selected) == 0 + assert len(result.notes) > 0 + + def test_all_gated_out(self, tmp_path): + """Pool where every remaining candidate fails safety gate.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + _write_panel_csv(pool, [ + _panel_row("CAND-A", "AAAAKKKFFF1", safety=0.1, rich_selectivity=0.2), + _panel_row("CAND-B", "BBBBLLLGGG2", safety=0.2, rich_selectivity=0.3), + ]) + + result = select_batch_2(pool, batch_1_ids=[], n=5) + assert len(result.selected) == 0 + assert result.n_after_safety_gate == 0 + assert len(result.notes) > 0 + + def test_batch_selection_has_reason(self, tmp_path): + """Each selected candidate has a selection_reason field.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + rows = [ + _panel_row(f"CAND-{i:03d}", f"AAAAKKKKFF{i}", ensemble=0.80 - i * 0.02) + for i in range(10) + ] + _write_panel_csv(pool, rows) + + result = select_batch_2(pool, batch_1_ids=[], n=5) + for c in result.selected: + assert "selection_reason" in c + assert isinstance(c["selection_reason"], str) + assert len(c["selection_reason"]) > 0 + + def test_selection_metadata_shape(self, tmp_path): + """BatchSelection dataclass has all expected fields.""" + from openamp_foundry.active_learning.selector import select_batch_2 + + pool = tmp_path / "pool.csv" + rows = [ + _panel_row(f"CAND-{i:03d}", f"AAAAKKKKFF{i}") for i in range(10) + ] + _write_panel_csv(pool, rows) + + result = select_batch_2(pool, batch_1_ids=[], n=3) + d = result.to_dict() + assert "version" in d + assert "n_requested" in d + assert "n_remaining" in d + assert "n_after_safety_gate" in d + assert "selected" in d + assert "probes_in_top_n" in d + assert "notes" in d + assert len(d["selected"]) == 3 + + def test_cli_select_batch(self, tmp_path): + """CLI select-batch exits 0 and produces valid JSON output.""" + pool = tmp_path / "pool.csv" + rows = [ + _panel_row(f"CAND-{i:03d}", f"AAAAKKKKFF{i}") for i in range(15) + ] + _write_panel_csv(pool, rows) + out = tmp_path / "batch2.json" + + result = subprocess.run( + [ + sys.executable, "-m", "openamp_foundry.cli", "select-batch", + "--candidates", str(pool), + "--batch-1-ids", "CAND-000,CAND-001", + "--n", "5", + "--out", str(out), + ], + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, f"CLI failed: {result.stderr}" + assert out.exists() + data = json.loads(out.read_text()) + assert len(data["selected"]) == 5 + assert data["n_requested"] == 5 + assert "version" in data + + def test_cli_rejects_missing_candidates(self, tmp_path): + """CLI exits 1 when candidates CSV does not exist.""" + out = tmp_path / "nonexistent.csv" + result = subprocess.run( + [ + sys.executable, "-m", "openamp_foundry.cli", "select-batch", + "--candidates", str(out), + "--batch-1-ids", "CAND-001", + "--out", str(tmp_path / "batch2.json"), + ], + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 1 + assert "error" in result.stdout