feat: Phase P P4 pilot batch safety clearance schema -- 4-screen safety gate blocks high-risk batches

docs/evidence/METRICS_CURRENT.md

@@ -1231,6 +1231,7 @@ Decoys score low on activity. Selective AMPs score moderately on both.
 
 | Date | Change | Author |
 |------|--------|--------|
+| 2026-07-10 | **Pilot batch safety clearance schema (Phase P P4):** PilotBatchSafetyClearanceEntry (13 fields: psc_id, bsp_id, pipeline_version, dual_use_risk_checked, novelty_verified, toxicity_screened, hemolysis_screened, max_safety_risk_tier, cleared_for_synthesis, rejection_ids, safety_notes, reviewer, dry_lab_only). All 4 safety screens mandatory. 8 validation rules: PSC-/BSP- prefixes, 4 screens all required True, risk tier in {low,moderate,high}, high-risk cannot be cleared (cleared_for_synthesis must be False), notes<=400 chars. 3 warnings: moderate-risk cleared, rejection_ids present, dry_lab_only clearance needs human review. 63 tests. CLI: openamp-foundry pilot-batch-safety-clearance-check. Safety constraint: max_safety_risk_tier=high cannot set cleared_for_synthesis=True. BASELINE 4847->4910. | OpenAMP Loop |
 | 2026-07-10 | **Batch outcome summary schema (Phase P P3):** BatchOutcomeSummaryEntry (12 fields: bos_id, pipeline_version, bsp_id, batch_id, candidates_proposed, candidates_tested, candidates_active, candidates_inactive, is_synthetic, outcome_notes, reviewer, dry_lab_only). 9 validation rules (BOS-/BSP- prefixes, count constraints, synthetic/real boundary: is_synthetic=True requires dry_lab_only=True, notes<=400 chars). 3 warnings (untested candidates, real results marked dry-lab-only, zero active). 63 tests. CLI: openamp-foundry batch-outcome-summary-check. Closes the BSP-to-outcomes feedback loop. BASELINE 4784->4847. | OpenAMP Loop |
 | 2026-07-10 | **Recalibration refusal record schema (Phase P P2):** RecalibrationRefusalEntry (10 fields, recalibration_refused must be True). 7 validation rules, 2 warnings. 63 tests. CLI: openamp-foundry recalibration-refusal-check. Complements CalibrationImprovementRecord (O3) with the rejection path. Corrected inflated BASELINE 4889->4784. | OpenAMP Loop |
 | 2026-07-09 | **Release checklist and gate validator (J1, starts Phase J):** `docs/governance/RELEASE_CHECKLIST.md` with structured checklist. `src/openamp_foundry/governance/release_gate.py` with `RELEASE_TYPES` (5), `UNIVERSAL_GATES` (7), `EXTRA_GATES_BY_TYPE`, `ReleaseGateResult`, `validate_release_gate()`. CLI `release-gate-check`. `make release-gate-check`. 18 tests. **Phase J (Governance and release maturity) started. 3478 total.** | OpenAMP Loop 109 |

docs/evidence/PILOT_BATCH_SAFETY_CLEARANCE_GUIDE.md

@@ -0,0 +1,72 @@
+# Pilot Batch Safety Clearance Guide
+
+## Purpose
+
+A Pilot Batch Safety Clearance (PSC) documents that a proposed batch (BSP) has been
+safety-reviewed before wet-lab synthesis is authorized.
+
+The PSC is a gate, not a rubber stamp. A batch with `max_safety_risk_tier="high"`
+**cannot be cleared** — `cleared_for_synthesis` must be `False` in that case.
+This constraint is enforced by the validator and cannot be overridden.
+
+## The four mandatory safety screens
+
+All four must be `True` for a valid PSC:
+
+| Screen | What it checks |
+|---|---|
+| `dual_use_risk_checked` | Risk that synthesized candidates could be repurposed for harm |
+| `novelty_verified` | Candidates are not re-synthesis of known hazardous sequences |
+| `toxicity_screened` | Computational toxicity prediction was run |
+| `hemolysis_screened` | Computational hemolysis prediction was run |
+
+These screens are required to be marked `True` — marking them `True` attests that
+the screen was performed. The PSC does not validate the quality of the screen.
+Qualified human review must verify that the screens were performed correctly.
+
+## Risk tier definitions
+
+| Tier | Meaning | Can clear? |
+|---|---|---|
+| `low` | No concerning signals from any screen | Yes |
+| `moderate` | One or more screens raised flags; individual review done | Yes (with warning) |
+| `high` | One or more candidates pose unacceptable risk | No |
+
+## Schema fields
+
+| Field | Type | Description |
+|---|---|---|
+| `psc_id` | str | Must start with `PSC-` |
+| `bsp_id` | str | Must start with `BSP-` |
+| `pipeline_version` | str | Pipeline version at review time |
+| `dual_use_risk_checked` | bool | Must be `True` |
+| `novelty_verified` | bool | Must be `True` |
+| `toxicity_screened` | bool | Must be `True` |
+| `hemolysis_screened` | bool | Must be `True` |
+| `max_safety_risk_tier` | str | `low`, `moderate`, or `high` |
+| `cleared_for_synthesis` | bool | Must be `False` if tier is `high` |
+| `rejection_ids` | List[str] | Candidate IDs rejected for safety |
+| `safety_notes` | str | Max 400 chars |
+| `reviewer` | str | Who performed or approved the safety review |
+| `dry_lab_only` | bool | `True` = computational screen only; `False` = wet-lab safety assay |
+
+## Honest boundaries
+
+- A PSC with `dry_lab_only=True` is a computational safety gate only. It is necessary
+  but not sufficient — qualified human review must precede actual synthesis.
+- `cleared_for_synthesis=True` means the pipeline's safety gate passed. It does not
+  mean synthesis is safe; it means no automated red flags were raised.
+- Safety screens check for known risk patterns. Novel sequences may pose risks not
+  captured by current screening methods.
+- The PSC does not validate that `bsp_id` references a real BSP; referential integrity
+  requires a separate audit step.
+
+## Warnings
+
+The validator emits warnings (not errors) for:
+
+- `max_safety_risk_tier="moderate"` with `cleared_for_synthesis=True` — ensure
+  moderate-risk candidates were individually reviewed.
+- Non-empty `rejection_ids` — lists the rejected candidates.
+- `dry_lab_only=True` with `cleared_for_synthesis=True` — computational clearance
+  requires additional qualified human review before synthesis.

docs/research/NEXT_100_PR_MAP.md

@@ -260,6 +260,7 @@ preventing the pipeline from selecting candidates under poor calibration.
 | P1 | Add batch selection proposal schema (complete). | Enforces calibration gate must pass before proposing a next batch; documents exploitation/exploration strategy. | B/D |
 | P2 | Add recalibration refusal record schema (complete). | Documents when recalibration was correctly rejected; prevents spurious recalibrations and creates audit trail. | B/D |
 | P3 | Add batch outcome summary schema (complete). | Closes the BSP→lab→outcomes feedback loop; enforces synthetic/real boundary at the batch level. | B/D |
+| P4 | Add pilot batch safety clearance schema (complete). | Safety gate before wet-lab synthesis: all 4 screens required, high-risk batches cannot be cleared. | B/D |
 
 ## Prioritization rule
 

src/openamp_foundry/cli/commands/reports.py

@@ -3666,3 +3666,33 @@ def _run_batch_outcome_summary_check(args) -> int:
         for warn in result.warnings:
             print(f"  WARN:  {warn}")
     return 0 if result.passed else 1
+
+
+def _run_pilot_batch_safety_clearance_check(args) -> int:
+    import json
+    from openamp_foundry.evidence.pilot_batch_safety_clearance import (
+        validate_pilot_batch_safety_clearance_dict,
+    )
+    if args.entry_json:
+        try:
+            data = json.loads(args.entry_json)
+        except json.JSONDecodeError as exc:
+            print(f"Error: invalid JSON: {exc}", file=__import__("sys").stderr)
+            return 1
+    else:
+        data = json.load(__import__("sys").stdin)
+    result = validate_pilot_batch_safety_clearance_dict(data)
+    if args.format == "json":
+        import dataclasses
+        print(json.dumps(dataclasses.asdict(result), indent=2))
+    else:
+        status = "PASS" if result.passed else "FAIL"
+        cleared = "CLEARED" if result.cleared_for_synthesis else "NOT-CLEARED"
+        print(f"[{status}] Pilot Batch Safety Clearance: {result.psc_id} [{cleared}]")
+        print(f"  BSP: {result.bsp_id}, risk tier: {result.max_safety_risk_tier}")
+        print(f"  Rejections: {result.rejection_count}")
+        for err in result.errors:
+            print(f"  ERROR: {err}")
+        for warn in result.warnings:
+            print(f"  WARN:  {warn}")
+    return 0 if result.passed else 1

src/openamp_foundry/cli/main.py

@@ -26,6 +26,7 @@
     _run_batch_selection_proposal_check,
     _run_recalibration_refusal_check,
     _run_batch_outcome_summary_check,
+    _run_pilot_batch_safety_clearance_check,
 )
 from openamp_foundry.cli.commands.gates import _run_gate_check, _run_release_gate_check
 from openamp_foundry.cli.commands.reports import _run_contribution_check
@@ -2383,6 +2384,14 @@ def build_parser() -> argparse.ArgumentParser:
     p_bos.add_argument("--format", choices=["text", "json"], default="text")
     p_bos.set_defaults(func=_run_batch_outcome_summary_check)
 
+    p_psc = sub.add_parser(
+        "pilot-batch-safety-clearance-check",
+        help="Validate pilot batch safety clearance entry",
+    )
+    p_psc.add_argument("--entry-json", default=None)
+    p_psc.add_argument("--format", choices=["text", "json"], default="text")
+    p_psc.set_defaults(func=_run_pilot_batch_safety_clearance_check)
+
     # ── Selection rationale check (Phase K K1) ───────────────────────
     src2 = sub.add_parser(
         "selection-rationale-check",
@@ -2767,6 +2776,9 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "batch-outcome-summary-check":
         return _run_batch_outcome_summary_check(args)
 
+    if args.command == "pilot-batch-safety-clearance-check":
+        return _run_pilot_batch_safety_clearance_check(args)
+
     parser.error("unknown command")
     return 2
 

src/openamp_foundry/evidence/pilot_batch_safety_clearance.py

@@ -0,0 +1,199 @@
+"""Pilot batch safety clearance schema — Phase P P4.
+
+Documents that a proposed batch has been safety-reviewed before wet-lab synthesis.
+This schema prevents the pipeline from clearing hazardous sequences for synthesis
+by requiring all four safety screens to complete and blocking high-risk batches.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import List
+
+PSC_PREFIX = "PSC-"
+BSP_PREFIX = "BSP-"
+VALID_RISK_TIERS = frozenset({"low", "moderate", "high"})
+SAFETY_NOTES_MAX_LENGTH = 400
+HIGH_RISK_TIER = "high"
+MODERATE_RISK_TIER = "moderate"
+
+
+@dataclass
+class PilotBatchSafetyClearanceEntry:
+    """Safety clearance record for a proposed batch before wet-lab synthesis."""
+
+    psc_id: str
+    bsp_id: str
+    pipeline_version: str
+    dual_use_risk_checked: bool
+    novelty_verified: bool
+    toxicity_screened: bool
+    hemolysis_screened: bool
+    max_safety_risk_tier: str
+    cleared_for_synthesis: bool
+    rejection_ids: List[str]
+    safety_notes: str
+    reviewer: str
+    dry_lab_only: bool = True
+
+
+@dataclass
+class PilotBatchSafetyClearanceResult:
+    psc_id: str
+    bsp_id: str
+    pipeline_version: str
+    max_safety_risk_tier: str
+    cleared_for_synthesis: bool
+    rejection_count: int
+    passed: bool
+    errors: List[str]
+    warnings: List[str]
+    dry_lab_only: bool = True
+
+
+def validate_pilot_batch_safety_clearance(
+    entry: PilotBatchSafetyClearanceEntry,
+) -> PilotBatchSafetyClearanceResult:
+    errors: List[str] = []
+    warnings: List[str] = []
+
+    if not entry.psc_id.startswith(PSC_PREFIX):
+        errors.append(
+            f"psc_id must start with '{PSC_PREFIX}', got '{entry.psc_id}'"
+        )
+
+    if not entry.bsp_id.startswith(BSP_PREFIX):
+        errors.append(
+            f"bsp_id must start with '{BSP_PREFIX}', got '{entry.bsp_id}'"
+        )
+
+    if not entry.dual_use_risk_checked:
+        errors.append(
+            "dual_use_risk_checked must be True — dual-use risk review is mandatory "
+            "before any batch is cleared for synthesis"
+        )
+
+    if not entry.novelty_verified:
+        errors.append(
+            "novelty_verified must be True — novelty verification is mandatory "
+            "to confirm candidates are not re-synthesis of known hazardous sequences"
+        )
+
+    if not entry.toxicity_screened:
+        errors.append(
+            "toxicity_screened must be True — toxicity screening is mandatory "
+            "before any batch is cleared for synthesis"
+        )
+
+    if not entry.hemolysis_screened:
+        errors.append(
+            "hemolysis_screened must be True — hemolysis screening is mandatory "
+            "before any batch is cleared for synthesis"
+        )
+
+    if entry.max_safety_risk_tier not in VALID_RISK_TIERS:
+        errors.append(
+            f"max_safety_risk_tier must be one of {sorted(VALID_RISK_TIERS)}, "
+            f"got '{entry.max_safety_risk_tier}'"
+        )
+
+    if entry.max_safety_risk_tier == HIGH_RISK_TIER and entry.cleared_for_synthesis:
+        errors.append(
+            "cleared_for_synthesis must be False when max_safety_risk_tier='high' — "
+            "high-risk batches cannot be cleared for wet-lab synthesis"
+        )
+
+    if len(entry.safety_notes) > SAFETY_NOTES_MAX_LENGTH:
+        errors.append(
+            f"safety_notes exceeds {SAFETY_NOTES_MAX_LENGTH} characters "
+            f"(got {len(entry.safety_notes)})"
+        )
+
+    # Warnings
+    if (
+        entry.max_safety_risk_tier == MODERATE_RISK_TIER
+        and entry.cleared_for_synthesis
+        and not errors
+    ):
+        warnings.append(
+            "max_safety_risk_tier is 'moderate' but batch is cleared — ensure "
+            "moderate-risk candidates have been individually reviewed and documented "
+            "in safety_notes or rejection_ids"
+        )
+
+    if entry.rejection_ids and not errors:
+        warnings.append(
+            f"{len(entry.rejection_ids)} candidate(s) rejected for safety: "
+            f"{entry.rejection_ids[:3]}{'...' if len(entry.rejection_ids) > 3 else ''}"
+        )
+
+    if entry.dry_lab_only and entry.cleared_for_synthesis and not errors:
+        warnings.append(
+            "dry_lab_only=True: this clearance is based on computational safety "
+            "screening only — wet-lab synthesis should receive additional "
+            "qualified human review before proceeding"
+        )
+
+    return PilotBatchSafetyClearanceResult(
+        psc_id=entry.psc_id,
+        bsp_id=entry.bsp_id,
+        pipeline_version=entry.pipeline_version,
+        max_safety_risk_tier=entry.max_safety_risk_tier,
+        cleared_for_synthesis=entry.cleared_for_synthesis,
+        rejection_count=len(entry.rejection_ids),
+        passed=len(errors) == 0,
+        errors=errors,
+        warnings=warnings,
+        dry_lab_only=entry.dry_lab_only,
+    )
+
+
+def validate_pilot_batch_safety_clearance_dict(
+    d: dict,
+) -> PilotBatchSafetyClearanceResult:
+    missing = []
+    for k in (
+        "psc_id",
+        "bsp_id",
+        "pipeline_version",
+        "dual_use_risk_checked",
+        "novelty_verified",
+        "toxicity_screened",
+        "hemolysis_screened",
+        "max_safety_risk_tier",
+        "cleared_for_synthesis",
+        "rejection_ids",
+        "safety_notes",
+        "reviewer",
+    ):
+        if k not in d:
+            missing.append(k)
+    if missing:
+        return PilotBatchSafetyClearanceResult(
+            psc_id=d.get("psc_id", ""),
+            bsp_id=d.get("bsp_id", ""),
+            pipeline_version=d.get("pipeline_version", ""),
+            max_safety_risk_tier=d.get("max_safety_risk_tier", ""),
+            cleared_for_synthesis=d.get("cleared_for_synthesis", False),
+            rejection_count=0,
+            passed=False,
+            errors=[f"missing required fields: {missing}"],
+            warnings=[],
+            dry_lab_only=d.get("dry_lab_only", True),
+        )
+    entry = PilotBatchSafetyClearanceEntry(
+        psc_id=d["psc_id"],
+        bsp_id=d["bsp_id"],
+        pipeline_version=d["pipeline_version"],
+        dual_use_risk_checked=bool(d["dual_use_risk_checked"]),
+        novelty_verified=bool(d["novelty_verified"]),
+        toxicity_screened=bool(d["toxicity_screened"]),
+        hemolysis_screened=bool(d["hemolysis_screened"]),
+        max_safety_risk_tier=d["max_safety_risk_tier"],
+        cleared_for_synthesis=bool(d["cleared_for_synthesis"]),
+        rejection_ids=list(d["rejection_ids"]),
+        safety_notes=d["safety_notes"],
+        reviewer=d["reviewer"],
+        dry_lab_only=bool(d.get("dry_lab_only", True)),
+    )
+    return validate_pilot_batch_safety_clearance(entry)