diff --git a/Makefile b/Makefile index 833b5d7e..74cb38fb 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: help demo test lint ci clean bench-leakage bench-multi-negatives bench-baseline bench-hidden-active bench-cluster-split bench-expert-ablation bench-expert-ablation-500 bench-selectivity bench-feature-decomp bench-gate bench-easy-baseline bench-charge-matched bench-order-dependent bench-precision-at-k bench-per-family bench-simulation-ablation bench-simulation-ablation-within-amp simulation-gate regenerate-all generate phase3 pilot validate-scoring validate-scoring-phase3 validate-scoring-strict external-pilot pilot-confident presynth-qc gold-standard diversity synthesis-order novelty-broad external-consensus questionnaire gate-check ip-report benchmark-card wave0-5-gate-check wave0-5-novelty-audit wave0-5-novelty-audit-v2 wave0-5-panel wave0-5-evidence wave0-5-fill-external wave0-5b-generate wave0-5b-filter recalibration-engine recalibration-engine-dry-run validate-policy-version generate-synthetic-lab-results bump-policy-version generate-review-packet failed-candidate-report +.PHONY: help demo test lint ci clean bench-leakage bench-multi-negatives bench-baseline bench-hidden-active bench-cluster-split bench-expert-ablation bench-expert-ablation-500 bench-selectivity bench-feature-decomp bench-gate bench-easy-baseline bench-charge-matched bench-order-dependent bench-precision-at-k bench-per-family bench-simulation-ablation bench-simulation-ablation-within-amp simulation-gate regenerate-all generate phase3 pilot validate-scoring validate-scoring-phase3 validate-scoring-strict external-pilot pilot-confident presynth-qc gold-standard diversity synthesis-order novelty-broad external-consensus questionnaire gate-check ip-report benchmark-card wave0-5-gate-check wave0-5-novelty-audit wave0-5-novelty-audit-v2 wave0-5-panel wave0-5-evidence wave0-5-fill-external wave0-5b-generate wave0-5b-filter recalibration-engine recalibration-engine-dry-run validate-policy-version generate-synthetic-lab-results bump-policy-version generate-review-packet failed-candidate-report safe-publication-filter PYTHON := $(shell [ -f .venv/bin/python ] && echo .venv/bin/python || echo python3) PYTEST := $(shell [ -f .venv/bin/pytest ] && echo .venv/bin/pytest || echo pytest) @@ -602,3 +602,9 @@ failed-candidate-report: --out-json outputs/failed_candidate_report.json \ --out-md outputs/failed_candidate_report.md \ --validate-rejection-codes + +safe-publication-filter: + PYTHONPATH=src $(PYTHON) scripts/safe_publication_filter.py \ + --input examples/safe_publication_filter_example_input.json \ + --out-json outputs/safe_publication_filter_result.json \ + --out-md outputs/safe_publication_filter_result.md diff --git a/docs/evidence/METRICS_CURRENT.md b/docs/evidence/METRICS_CURRENT.md index c952a8bf..3ef04f87 100644 --- a/docs/evidence/METRICS_CURRENT.md +++ b/docs/evidence/METRICS_CURRENT.md @@ -5,7 +5,9 @@ Machine-readable snapshot: `outputs/metrics_snapshot.json` regenerated with `mak > **Purpose:** One authoritative table of current pipeline metrics. If any doc disagrees > with this file, this file wins. Updated whenever benchmark/benchmark config changes. > -> **Last updated:** 2026-07-09 (Phase F report generator — v0.5.75) +> **Last updated:** 2026-07-09 (Phase F safe-publication filter — v0.5.76) +> **New in v0.5.76:** Safe-publication filter (F5) — CLI that reads candidate panel data with safety metadata and checks each candidate against publication safety constraints before external release. FAIL-CLOSED: defaults to reject unless all checks pass. Checks: dry_lab_only (must be true), proof_ladder_level (≤4), toxicity_screened, hemolysis_screened, dual_use_reviewed (all must be true). Produces structured JSON filter result and human-readable Markdown summary. 33 tests. 2758 total. +> > **New in v0.5.75:** Failed-candidate report generator (F4) — CLI that reads candidate rejection data + taxonomy, produces structured JSON report with summary aggregation (by category, severity, stage, evidence impact) and human-readable Markdown summary. 26 tests. 2724 total. > > **New in v0.5.74:** Rejection reason taxonomy schema (F3) — machine-readable taxonomy with severity, evidence_impact, and applies_at_stage fields. Comprehensive REJECTION_TAXONOMY.md doc. 19 tests. 2698 total. @@ -60,7 +62,7 @@ Machine-readable snapshot: `outputs/metrics_snapshot.json` regenerated with `mak > **New in v0.5.31:** Added dipeptide-order features for sequence-order awareness. `dipeptide_order_score` achieves AUROC 0.7861 on AMP-vs-scrambled discrimination — the strongest order-dependent feature in the pipeline. Only 7/31 features survive scrambling (amphipathicity/helix-wheel + dipeptide). All composition features are purely position-independent (exactly 0.5000 AUROC on scrambled test). > **New in v0.5.30:** Easy baseline benchmark added — charge density alone (AUROC 0.8166) outperforms the full pipeline ensemble (0.7792) on AMP-vs-Swiss-Prot-decoy discrimination. Honest finding documented: expected because pipeline optimizes for safety, not raw discrimination. > **New in v0.5.29:** Expanded benchmark to 500 AMPs + 500 composition-matched decoys (n=1000). AUROC 0.7792 (CI₉₅: 0.7505–0.8065) confirms signal generalizes. Cluster-aware CI: 0.746–0.8102. Representative AUROC: 0.778. Standard benchmark (n=191) retained for backward comparison. -> **Pipeline version:** v0.5.75 +> **Pipeline version:** v0.5.76 > **Branch:** main --- diff --git a/docs/research/ROADMAP.md b/docs/research/ROADMAP.md index bea5e03c..e6a89ea0 100644 --- a/docs/research/ROADMAP.md +++ b/docs/research/ROADMAP.md @@ -1,5 +1,49 @@ # Roadmap +## v0.5.76 — Loop 75: Phase F F5 — Safe-Publication Filter ✓ (2026-07-09) + +Safe-publication filter: CLI that reads a candidate panel with safety metadata +and checks each candidate against publication safety constraints before external +release. FAIL-CLOSED: defaults to reject unless all checks pass. + +Changes: +- `scripts/safe_publication_filter.py` (F5) — Standalone CLI that loads candidate + panel data with publication safety metadata and applies global-level checks + (dry_lab_only must be true, proof_ladder_level ≤ 4, toxicity_screened, + hemolysis_screened, dual_use_reviewed all true) plus per-candidate checks + (proof_ladder_level ≤ 4). Produces structured JSON with filter metadata, + global checks, per-candidate results, and summary. Exit code 0 if all + checks pass, 1 if any fail (FAIL-CLOSED), 2 for input errors. +- `schemas/safe_publication_filter.schema.json` (F5) — JSON Schema Draft 2020-12 + for the filter result output. Validates filter_metadata, global_checks + (3 required sub-check groups), per-candidate results, summary counts, and + caveats. +- `examples/safe_publication_filter_example_input.json` (F5) — Toy input + with 4 candidates spanning PASS (pll=2, pll=1), WARN (pll=4 at boundary), + and FAIL (pll=5 too high). Clearly marked EXAMPLE. +- `Makefile` — Added `safe-publication-filter` target. +- `tests/evidence/test_safe_publication_filter.py` — 33 tests covering: + global check success/failure paths (dry_lab_only, proof_ladder_level, + all safety checks individually), per-candidate checks (valid, missing, + too-high, boundary/warn), mixed candidate results, filter result summary + counts, overall PASS/FAIL logic, markdown output sections, input loading + (valid/missing/invalid JSON), CLI exit codes (0, 1, 2), CLI output + (JSON + Markdown), caveat generation, metadata structure, and example + input round-trip validation. + +Honest boundaries: +- The filter checks structural publication safety constraints, not scientific + validity. A PASS result does not mean the candidate is biologically active + or safe. +- dry_lab_only attestation is a safety-constraint field; it relies on the + generator's honesty. +- Proof ladder levels are self-reported by the pipeline; independent + verification is recommended. +- Safety checks (toxicity, hemolysis, dual-use) are computational assessments + and require qualified human review. +- A FAIL result does not mean the candidate is biologically inactive — it + means publication safety checks are not satisfied. + ## v0.5.75 — Loop 74: Phase F F4 — Failed-Candidate Report Generator ✓ (2026-07-09) Failed-candidate report generator: CLI that reads candidate rejection data validated diff --git a/examples/safe_publication_filter_example_input.json b/examples/safe_publication_filter_example_input.json new file mode 100644 index 00000000..951ac30a --- /dev/null +++ b/examples/safe_publication_filter_example_input.json @@ -0,0 +1,39 @@ +{ + "source": "wave0.5-example", + "pipeline_version": "v0.5.75", + "report_type": "safe_publication_input", + "description": "EXAMPLE safe-publication filter input. NOT a real candidate panel — illustrative only.", + "dry_lab_only": true, + "proof_ladder_level": 2, + "safety_checks": { + "toxicity_screened": true, + "hemolysis_screened": true, + "dual_use_reviewed": true + }, + "candidates": [ + { + "candidate_id": "EXAMPLE-CAND-001", + "sequence": "ACDEFGHIKLMNPQRSTVWY", + "ensemble_score": 0.85, + "proof_ladder_level": 2 + }, + { + "candidate_id": "EXAMPLE-CAND-002", + "sequence": "KLAKLAKKLAKLAK", + "ensemble_score": 0.80, + "proof_ladder_level": 4 + }, + { + "candidate_id": "EXAMPLE-CAND-003", + "sequence": "GIGKFLHSAKKWGKAFVGEIMNS", + "ensemble_score": 0.78, + "proof_ladder_level": 5 + }, + { + "candidate_id": "EXAMPLE-CAND-004", + "sequence": "RRWQWRMKKLG", + "ensemble_score": 0.72, + "proof_ladder_level": 1 + } + ] +} diff --git a/schemas/safe_publication_filter.schema.json b/schemas/safe_publication_filter.schema.json new file mode 100644 index 00000000..2cb7a814 --- /dev/null +++ b/schemas/safe_publication_filter.schema.json @@ -0,0 +1,163 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://openamp-foundry.org/schemas/safe_publication_filter.schema.json", + "title": "OpenAMP Safe Publication Filter Result", + "description": "JSON Schema for the safe-publication filter result. Validates the output of scripts/safe_publication_filter.py before any external release. FAIL-CLOSED: the filter defaults to reject unless all publication safety checks pass.", + "type": "object", + "required": [ + "filter_metadata", + "global_checks", + "global_pass", + "candidates", + "summary", + "caveats" + ], + "properties": { + "filter_metadata": { + "type": "object", + "required": [ + "generated_at", + "filter_type", + "filter_version", + "input_source", + "pipeline_version" + ], + "properties": { + "generated_at": { + "type": "string", + "format": "date-time", + "description": "ISO 8601 timestamp when the filter was run" + }, + "filter_type": { + "type": "string", + "enum": ["safe_publication_filter"], + "description": "Filter type identifier" + }, + "filter_version": { + "type": "string", + "pattern": "^\\d+\\.\\d+\\.\\d+$", + "description": "Filter schema version (semver)" + }, + "input_source": { + "type": "string", + "description": "Source identifier of the input data" + }, + "pipeline_version": { + "type": "string", + "description": "OpenAMP pipeline version" + } + } + }, + "global_checks": { + "type": "object", + "description": "Global-level publication safety checks applied to the entire batch", + "required": ["dry_lab_only", "global_proof_ladder_level", "safety_checks"], + "properties": { + "dry_lab_only": { + "type": "object", + "required": ["required", "actual", "passed", "description"], + "properties": { + "required": { "type": "boolean", "enum": [true] }, + "actual": { "type": "boolean" }, + "passed": { "type": "boolean" }, + "description": { "type": "string" }, + "reason": { "type": "string" } + } + }, + "global_proof_ladder_level": { + "type": "object", + "required": ["max_allowed", "actual", "passed", "description"], + "properties": { + "required": { "type": "string" }, + "max_allowed": { "type": "integer", "minimum": 1, "maximum": 8 }, + "actual": { "type": "integer", "minimum": 1, "maximum": 8 }, + "passed": { "type": "boolean" }, + "description": { "type": "string" }, + "reason": { "type": "string" } + } + }, + "safety_checks": { + "type": "object", + "required": ["all_passed", "checks"], + "properties": { + "all_passed": { "type": "boolean" }, + "checks": { + "type": "object", + "minProperties": 3, + "additionalProperties": { + "type": "object", + "required": ["required", "actual", "passed"], + "properties": { + "required": { "type": "boolean", "enum": [true] }, + "actual": { "type": "boolean" }, + "passed": { "type": "boolean" }, + "reason": { "type": "string" } + } + } + } + } + } + } + }, + "global_pass": { + "type": "boolean", + "description": "Whether all global-level checks passed. If false, the filter FAILS regardless of per-candidate results." + }, + "candidates": { + "type": "array", + "items": { + "type": "object", + "required": ["candidate_id", "overall", "checks"], + "properties": { + "candidate_id": { "type": "string" }, + "overall": { + "type": "string", + "enum": ["PASS", "FAIL", "WARN"], + "description": "PASS: all checks pass, FAIL: any check fails, WARN: passes but at boundary" + }, + "checks": { + "type": "object", + "minProperties": 1, + "additionalProperties": { + "type": "object", + "required": ["value", "max_allowed", "passed"], + "properties": { + "value": { "type": ["integer", "null"] }, + "max_allowed": { "type": "integer" }, + "passed": { "type": "boolean" }, + "reason": { "type": "string" } + } + } + }, + "issues": { + "type": "array", + "items": { "type": "string" } + } + } + } + }, + "summary": { + "type": "object", + "required": ["total_candidates", "passed", "failed", "overall_filter_pass"], + "properties": { + "total_candidates": { "type": "integer", "minimum": 0 }, + "passed": { "type": "integer", "minimum": 0 }, + "failed": { "type": "integer", "minimum": 0 }, + "warned": { "type": "integer", "minimum": 0 }, + "overall_filter_pass": { + "type": "boolean", + "description": "True only if ALL global checks pass AND NO candidates FAIL. FAIL-CLOSED." + } + } + }, + "caveats": { + "type": "array", + "minItems": 1, + "items": { "type": "string" } + }, + "dry_lab_only": { + "type": "boolean", + "description": "Copied from input — all outputs are dry-lab computational hypotheses, not biological proof." + } + } +} diff --git a/scripts/safe_publication_filter.py b/scripts/safe_publication_filter.py new file mode 100644 index 00000000..71225f06 --- /dev/null +++ b/scripts/safe_publication_filter.py @@ -0,0 +1,437 @@ +"""Safe-publication filter for candidate panels and negative-result reports. + +Fail-closed: default reject unless ALL checks pass. + +Checks: + - dry_lab_only must be true + - global proof_ladder_level must be <= 4 + - toxicity_screened must be true + - hemolysis_screened must be true + - dual_use_reviewed must be true + - per-candidate proof_ladder_level must be <= 4 + +Usage: + python scripts/safe_publication_filter.py \ + --input examples/safe_publication_filter_example_input.json \ + --out-json outputs/safe_publication_filter_result.json + + python scripts/safe_publication_filter.py \ + --input examples/safe_publication_filter_example_input.json + +Exit codes: + 0 ALL PASS — publication safety checks satisfied + 1 FAIL — one or more checks failed (fail-closed) + 2 INPUT ERROR — missing file, invalid JSON, or bad data +""" +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +def load_input(input_path: str | Path) -> dict[str, Any]: + """Load and validate input JSON file.""" + p = Path(input_path) + if not p.exists(): + raise FileNotFoundError(f"Input file not found: {p}") + try: + with p.open("r") as f: + return json.load(f) + except json.JSONDecodeError as e: + msg = f"Invalid JSON in {p}: {e}" + raise ValueError(msg) from e + + +def run_global_checks(data: dict[str, Any]) -> dict[str, Any]: + """Run global-level publication safety checks. + + Returns {checks: dict of check results, all_passed: bool}. + FAIL-CLOSED: any missing or false check fails the global gate. + """ + checks: dict[str, Any] = {} + all_passed = True + + # 1. dry_lab_only must be True + dry_lab = data.get("dry_lab_only", False) + dry_ok = dry_lab is True + if not dry_ok: + all_passed = False + dry_result: dict[str, Any] = { + "required": True, + "actual": dry_lab, + "passed": dry_ok, + "description": "dry_lab_only must be true — computational outputs are hypotheses, not biological proof", + } + if not dry_ok: + dry_result["reason"] = ( + f"dry_lab_only is {dry_lab!r}, must be true. " + "Publication-ready packets must explicitly acknowledge dry-lab-only status." + ) + checks["dry_lab_only"] = dry_result + + # 2. Global proof_ladder_level must be <= 4 + max_allowed = 4 + global_pll = data.get("proof_ladder_level", None) + pll_ok = isinstance(global_pll, (int, float)) and global_pll <= max_allowed + if not pll_ok: + all_passed = False + pll_result: dict[str, Any] = { + "required": f"\u2264 {max_allowed}", + "max_allowed": max_allowed, + "actual": global_pll if isinstance(global_pll, (int, float)) else None, + "passed": pll_ok, + "description": "Global proof ladder level must be \u2264 4 — levels \u2265 5 require biological validation data", + } + if not pll_ok: + if global_pll is None: + pll_result["reason"] = "proof_ladder_level is not specified in input." + else: + pll_result["reason"] = ( + f"proof_ladder_level is {global_pll}, exceeds max allowed ({max_allowed})." + ) + checks["global_proof_ladder_level"] = pll_result + + # 3. Safety sub-checks + safety = data.get("safety_checks", {}) + safety_check_names = ["toxicity_screened", "hemolysis_screened", "dual_use_reviewed"] + safety_results: dict[str, Any] = {} + safety_all_ok = True + for name in safety_check_names: + actual = safety.get(name, False) + ok = actual is True + if not ok: + safety_all_ok = False + all_passed = False + sr: dict[str, Any] = { + "required": True, + "actual": actual if isinstance(actual, bool) else False, + "passed": ok, + } + if not ok: + sr["reason"] = f"{name} is {actual!r}, must be true." + safety_results[name] = sr + + checks["safety_checks"] = { + "all_passed": safety_all_ok, + "checks": safety_results, + } + + return {"all_passed": all_passed, "checks": checks} + + +def run_candidate_checks(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Run per-candidate publication safety checks. + + Checks: + - proof_ladder_level must be <= 4 (PASS), == 4 → WARN, > 4 or missing → FAIL + """ + results: list[dict[str, Any]] = [] + max_allowed_pll = 4 + + for c in candidates: + cid = c.get("candidate_id", "?") + candidate_pll = c.get("proof_ladder_level", None) + + check_results: dict[str, Any] = {} + issues: list[str] = [] + + # proof_ladder_level check + if candidate_pll is None: + check_results["proof_ladder_level"] = { + "value": None, + "max_allowed": max_allowed_pll, + "passed": False, + "reason": "proof_ladder_level not specified for candidate", + } + issues.append("PROOF_LADDER_MISSING") + elif not isinstance(candidate_pll, (int, float)): + check_results["proof_ladder_level"] = { + "value": candidate_pll, + "max_allowed": max_allowed_pll, + "passed": False, + "reason": f"proof_ladder_level must be an integer, got {type(candidate_pll).__name__}", + } + issues.append("PROOF_LADDER_INVALID") + elif candidate_pll > max_allowed_pll: + check_results["proof_ladder_level"] = { + "value": candidate_pll, + "max_allowed": max_allowed_pll, + "passed": False, + "reason": f"proof_ladder_level {candidate_pll} exceeds max allowed {max_allowed_pll}", + } + issues.append("PROOF_LADDER_EXCEEDED") + else: + check_results["proof_ladder_level"] = { + "value": candidate_pll, + "max_allowed": max_allowed_pll, + "passed": True, + } + + # Determine overall: PASS if all pass, WARN if at boundary, FAIL otherwise + pll_check = check_results["proof_ladder_level"] + if not pll_check["passed"]: + overall = "FAIL" + elif isinstance(candidate_pll, (int, float)) and candidate_pll == max_allowed_pll: + overall = "WARN" + else: + overall = "PASS" + + result: dict[str, Any] = { + "candidate_id": cid, + "overall": overall, + "checks": check_results, + } + if issues: + result["issues"] = issues + results.append(result) + + return results + + +def build_filter_result( + input_data: dict[str, Any], + global_results: dict[str, Any], + candidate_results: list[dict[str, Any]], +) -> dict[str, Any]: + """Build the structured safe-publication filter result dict.""" + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + passed = sum(1 for c in candidate_results if c["overall"] == "PASS") + warned = sum(1 for c in candidate_results if c["overall"] == "WARN") + failed = sum(1 for c in candidate_results if c["overall"] == "FAIL") + total = len(candidate_results) + + global_pass = global_results["all_passed"] + overall_filter_pass = global_pass and failed == 0 + + return { + "filter_metadata": { + "generated_at": now, + "filter_type": "safe_publication_filter", + "filter_version": "1.0.0", + "input_source": input_data.get("source", "unknown"), + "pipeline_version": input_data.get("pipeline_version", "unknown"), + }, + "global_checks": global_results["checks"], + "global_pass": global_pass, + "candidates": candidate_results, + "summary": { + "total_candidates": total, + "passed": passed, + "failed": failed, + "warned": warned, + "overall_filter_pass": overall_filter_pass, + }, + "caveats": [ + "Computational outputs are hypotheses and review aids. They are not biological proof.", + "This filter checks structural publication safety constraints, not scientific validity.", + "A PASS result does not mean the candidate is biologically active or safe.", + "A FAIL result does not mean the candidate is biologically inactive — it means publication safety checks are not satisfied.", + "Proof ladder levels are self-reported by the pipeline; independent verification is recommended.", + "Safety checks (toxicity, hemolysis, dual-use) are computational assessments and require qualified human review.", + "dry_lab_only attestation is a safety-constraint field; it relies on the generator's honesty.", + ], + "dry_lab_only": input_data.get("dry_lab_only", False), + } + + +def build_markdown_result(result: dict[str, Any]) -> str: + """Generate a Markdown summary from the filter result.""" + meta = result["filter_metadata"] + summary = result["summary"] + lines: list[str] = [] + + lines.append("# Safe Publication Filter Result") + lines.append("") + lines.append("> **FAIL-CLOSED:** Default reject unless all checks pass. " + "Computational outputs are hypotheses and review aids. They are not biological proof.") + lines.append("") + + lines.append("## Filter Metadata") + lines.append("") + lines.append("| Field | Value |") + lines.append("|-------|-------|") + lines.append(f"| Generated at | {meta['generated_at']} |") + lines.append(f"| Input source | {meta['input_source']} |") + lines.append(f"| Pipeline version | {meta['pipeline_version']} |") + lines.append(f"| Filter version | {meta['filter_version']} |") + lines.append("") + + lines.append("## Global Checks") + lines.append("") + global_pass = result["global_pass"] + status = "PASS" if global_pass else "FAIL" + lines.append(f"**Overall global check status: {status}**") + lines.append("") + + gc = result["global_checks"] + + dry = gc.get("dry_lab_only", {}) + lines.append(f"**dry_lab_only:** {'✅ PASS' if dry.get('passed') else '❌ FAIL'}") + lines.append(f"- Required: {dry.get('required')}, Actual: {dry.get('actual')}") + if dry.get("reason"): + lines.append(f"- Reason: {dry['reason']}") + lines.append("") + + pll = gc.get("global_proof_ladder_level", {}) + lines.append(f"**Global proof_ladder_level:** {'✅ PASS' if pll.get('passed') else '❌ FAIL'}") + lines.append(f"- Max allowed: {pll.get('max_allowed')}, Actual: {pll.get('actual')}") + if pll.get("reason"): + lines.append(f"- Reason: {pll['reason']}") + lines.append("") + + sc = gc.get("safety_checks", {}) + lines.append(f"**Safety checks:** {'✅ ALL PASS' if sc.get('all_passed') else '❌ SOME FAILED'}") + for name, sr in sc.get("checks", {}).items(): + icon = "✅" if sr.get("passed") else "❌" + lines.append(f" - {icon} {name}: {sr.get('actual')}") + if sr.get("reason"): + lines.append(f" - Reason: {sr['reason']}") + lines.append("") + + lines.append("## Summary") + lines.append("") + lines.append("| Metric | Value |") + lines.append("|--------|-------|") + lines.append(f"| Total candidates | {summary['total_candidates']} |") + lines.append(f"| Passed | {summary['passed']} |") + lines.append(f"| Warned | {summary['warned']} |") + lines.append(f"| Failed | {summary['failed']} |") + pass_status = "PASS" if summary["overall_filter_pass"] else "FAIL" + lines.append(f"| **Overall filter verdict** | **{pass_status}** |") + lines.append("") + + if result["candidates"]: + lines.append("## Per-Candidate Results") + lines.append("") + lines.append("| Candidate | Overall | Proof Ladder Level | Issues |") + lines.append("|-----------|---------|-------------------|--------|") + for c in result["candidates"]: + pll_check = c["checks"].get("proof_ladder_level", {}) + pll_val = pll_check.get("value", "?") + icon = {"PASS": "✅", "WARN": "⚠️", "FAIL": "❌"}.get(c["overall"], "?") + issues_str = ", ".join(c.get("issues", [])) if c.get("issues") else "" + lines.append(f"| {c['candidate_id']} | {icon} {c['overall']} | {pll_val} | {issues_str} |") + lines.append("") + + lines.append("## Caveats") + lines.append("") + for caveat in result.get("caveats", []): + lines.append(f"- {caveat}") + lines.append("") + lines.append("---") + lines.append(f"*Generated by `scripts/safe_publication_filter.py` — " + f"`{meta['pipeline_version']}`*") + + return "\n".join(lines) + + +def validate_schema(result: dict[str, Any], schema_path: str | Path | None = None) -> bool: + """Validate filter result against the schema file.""" + schema_p = Path(schema_path) if schema_path else ( + Path(__file__).resolve().parent.parent / "schemas" / "safe_publication_filter.schema.json" + ) + if not schema_p.exists(): + raise FileNotFoundError(f"Schema not found: {schema_p}") + try: + import jsonschema + except ImportError: + return True # skip validation if jsonschema not available + with schema_p.open("r") as f: + schema = json.load(f) + try: + jsonschema.validate(result, schema) + return True + except jsonschema.ValidationError as e: + raise ValueError(f"Schema validation failed: {e}") from e + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Safe-publication filter — checks candidate panels against " + "publication safety constraints before external release.", + ) + parser.add_argument( + "--input", required=True, type=Path, + help="Path to input JSON file (candidate panel with safety metadata)", + ) + parser.add_argument( + "--out-json", type=Path, default=None, + help="Path to write JSON filter result (default: not written)", + ) + parser.add_argument( + "--out-md", type=Path, default=None, + help="Path to write Markdown summary (default: not written)", + ) + parser.add_argument( + "--validate-schema", action="store_true", + help="Validate filter result against schema (requires jsonschema)", + ) + parser.add_argument( + "--pipeline-version", default=None, + help="Override pipeline version (default: from input file metadata)", + ) + + args = parser.parse_args(argv) + + # Load input + try: + input_data = load_input(args.input) + except (FileNotFoundError, ValueError) as e: + print(f"Error loading input: {e}", file=sys.stderr) + return 2 + + candidates = input_data.get("candidates", []) + if not candidates: + print("Error: No candidates in input file", file=sys.stderr) + return 2 + + pipeline_version = args.pipeline_version or input_data.get( + "pipeline_version", "unknown" + ) + input_data.setdefault("pipeline_version", pipeline_version) + + # Run filter + global_results = run_global_checks(input_data) + candidate_results = run_candidate_checks(candidates) + + result = build_filter_result(input_data, global_results, candidate_results) + + # Validate schema if requested + if args.validate_schema: + try: + validate_schema(result) + except (FileNotFoundError, ValueError) as e: + print(f"Schema validation error: {e}", file=sys.stderr) + return 2 + + # Write JSON output + if args.out_json: + args.out_json.parent.mkdir(parents=True, exist_ok=True) + args.out_json.write_text( + json.dumps(result, indent=2, ensure_ascii=False) + "\n" + ) + print(f"Wrote filter result to {args.out_json}") + + # Write Markdown output + if args.out_md: + args.out_md.parent.mkdir(parents=True, exist_ok=True) + md_content = build_markdown_result(result) + args.out_md.write_text(md_content) + print(f"Wrote Markdown summary to {args.out_md}") + + if not args.out_json and not args.out_md: + print(json.dumps(result, indent=2, ensure_ascii=False)) + + # Exit code: 0 = all pass, 1 = any fail (FAIL-CLOSED) + if result["summary"]["overall_filter_pass"]: + return 0 + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/evidence/test_safe_publication_filter.py b/tests/evidence/test_safe_publication_filter.py new file mode 100644 index 00000000..18486707 --- /dev/null +++ b/tests/evidence/test_safe_publication_filter.py @@ -0,0 +1,452 @@ +"""Tests for the safe-publication filter (F5).""" +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from scripts.safe_publication_filter import ( + build_filter_result, + build_markdown_result, + load_input, + run_candidate_checks, + run_global_checks, +) + +PYTHON = sys.executable +_REPO_DIR = Path(__file__).parents[2] + + +def _default_input(overrides: dict | None = None) -> dict: + data = { + "source": "test-batch", + "pipeline_version": "v0.5.75-test", + "report_type": "safe_publication_input", + "dry_lab_only": True, + "proof_ladder_level": 2, + "safety_checks": { + "toxicity_screened": True, + "hemolysis_screened": True, + "dual_use_reviewed": True, + }, + "candidates": [ + { + "candidate_id": "CAND-001", + "sequence": "ACDEFGHIKLMNPQRSTVWY", + "ensemble_score": 0.85, + "proof_ladder_level": 2, + }, + { + "candidate_id": "CAND-002", + "sequence": "KLAKLAKKLAKLAK", + "ensemble_score": 0.80, + "proof_ladder_level": 1, + }, + ], + } + if overrides: + data.update(overrides) + return data + + +class TestSafePublicationFilter: + def test_global_checks_all_pass(self): + """All global checks pass with valid input.""" + result = run_global_checks(_default_input()) + assert result["all_passed"] is True + checks = result["checks"] + assert checks["dry_lab_only"]["passed"] is True + assert checks["global_proof_ladder_level"]["passed"] is True + assert checks["safety_checks"]["all_passed"] is True + + def test_global_dry_lab_only_false_fails(self): + """dry_lab_only=false fails global checks.""" + result = run_global_checks(_default_input({"dry_lab_only": False})) + assert result["all_passed"] is False + assert result["checks"]["dry_lab_only"]["passed"] is False + + def test_global_dry_lab_only_missing_fails(self): + """Missing dry_lab_only defaults to False and fails.""" + result = run_global_checks(_default_input({"dry_lab_only": False})) + assert result["all_passed"] is False + + def test_global_proof_ladder_too_high_fails(self): + """proof_ladder_level > 4 fails global checks.""" + result = run_global_checks(_default_input({"proof_ladder_level": 5})) + assert result["all_passed"] is False + assert result["checks"]["global_proof_ladder_level"]["passed"] is False + assert "exceeds" in result["checks"]["global_proof_ladder_level"].get("reason", "") + + def test_global_proof_ladder_missing_fails(self): + """Missing proof_ladder_level fails checks.""" + result = run_global_checks(_default_input({"proof_ladder_level": None})) + assert result["all_passed"] is False + assert result["checks"]["global_proof_ladder_level"]["passed"] is False + + def test_global_toxicity_not_screened_fails(self): + """toxicity_screened=false fails safety checks.""" + result = run_global_checks(_default_input({ + "safety_checks": { + "toxicity_screened": False, + "hemolysis_screened": True, + "dual_use_reviewed": True, + } + })) + assert result["all_passed"] is False + sc = result["checks"]["safety_checks"] + assert sc["all_passed"] is False + assert sc["checks"]["toxicity_screened"]["passed"] is False + + def test_global_hemolysis_not_screened_fails(self): + """hemolysis_screened=false fails safety checks.""" + result = run_global_checks(_default_input({ + "safety_checks": { + "toxicity_screened": True, + "hemolysis_screened": False, + "dual_use_reviewed": True, + } + })) + assert result["all_passed"] is False + sc = result["checks"]["safety_checks"] + assert sc["all_passed"] is False + assert sc["checks"]["hemolysis_screened"]["passed"] is False + + def test_global_dual_use_not_reviewed_fails(self): + """dual_use_reviewed=false fails safety checks.""" + result = run_global_checks(_default_input({ + "safety_checks": { + "toxicity_screened": True, + "hemolysis_screened": True, + "dual_use_reviewed": False, + } + })) + assert result["all_passed"] is False + sc = result["checks"]["safety_checks"] + assert sc["all_passed"] is False + assert sc["checks"]["dual_use_reviewed"]["passed"] is False + + def test_global_safety_missing_all_fails(self): + """Missing safety_checks section fails all three.""" + result = run_global_checks(_default_input({"safety_checks": {}})) + assert result["all_passed"] is False + sc = result["checks"]["safety_checks"] + assert sc["all_passed"] is False + for name in ["toxicity_screened", "hemolysis_screened", "dual_use_reviewed"]: + assert sc["checks"][name]["passed"] is False + + def test_candidate_pll_too_high_fails(self): + """Candidate with proof_ladder_level > 4 fails.""" + candidates = [{ + "candidate_id": "CAND-001", + "proof_ladder_level": 5, + }] + results = run_candidate_checks(candidates) + assert results[0]["overall"] == "FAIL" + assert "PROOF_LADDER_EXCEEDED" in results[0].get("issues", []) + + def test_candidate_pll_missing_fails(self): + """Candidate missing proof_ladder_level fails.""" + candidates = [{"candidate_id": "CAND-001"}] + results = run_candidate_checks(candidates) + assert results[0]["overall"] == "FAIL" + assert "PROOF_LADDER_MISSING" in results[0].get("issues", []) + + def test_candidate_pll_at_boundary_warns(self): + """Candidate with proof_ladder_level == 4 returns WARN.""" + candidates = [{ + "candidate_id": "CAND-001", + "proof_ladder_level": 4, + }] + results = run_candidate_checks(candidates) + assert results[0]["overall"] == "WARN" + assert results[0]["checks"]["proof_ladder_level"]["passed"] is True + + def test_candidate_pll_valid_passes(self): + """Candidate with valid proof_ladder_level passes.""" + candidates = [{ + "candidate_id": "CAND-001", + "proof_ladder_level": 2, + }] + results = run_candidate_checks(candidates) + assert results[0]["overall"] == "PASS" + assert results[0]["checks"]["proof_ladder_level"]["passed"] is True + + def test_candidate_multiple_mixed_results(self): + """Multiple candidates with mixed levels produce correct results.""" + candidates = [ + {"candidate_id": "C-001", "proof_ladder_level": 2}, + {"candidate_id": "C-002", "proof_ladder_level": 5}, + {"candidate_id": "C-003", "proof_ladder_level": 4}, + ] + results = run_candidate_checks(candidates) + statuses = {r["candidate_id"]: r["overall"] for r in results} + assert statuses["C-001"] == "PASS" + assert statuses["C-002"] == "FAIL" + assert statuses["C-003"] == "WARN" + + def test_build_filter_result_summary(self): + """Filter result summary counts are correct.""" + input_data = _default_input() + global_results = run_global_checks(input_data) + candidate_results = run_candidate_checks(input_data["candidates"]) + result = build_filter_result(input_data, global_results, candidate_results) + s = result["summary"] + candidates = input_data["candidates"] + assert s["total_candidates"] == len(candidates) + assert s["passed"] + s["warned"] + s["failed"] == s["total_candidates"] + assert s["overall_filter_pass"] is True + + def test_build_filter_result_summary_with_fails(self): + """Overall filter pass is False when candidates fail.""" + input_data = _default_input({ + "dry_lab_only": True, + "proof_ladder_level": 2, + "safety_checks": {"toxicity_screened": True, "hemolysis_screened": True, "dual_use_reviewed": True}, + "candidates": [ + {"candidate_id": "C-001", "proof_ladder_level": 2}, + {"candidate_id": "C-002", "proof_ladder_level": 5}, + ], + }) + global_results = run_global_checks(input_data) + candidate_results = run_candidate_checks(input_data["candidates"]) + result = build_filter_result(input_data, global_results, candidate_results) + assert result["summary"]["overall_filter_pass"] is False + assert result["summary"]["failed"] == 1 + + def test_global_fail_makes_overall_fail_even_with_pass_candidates(self): + """Global check failure makes overall_filter_pass False even with passing candidates.""" + input_data = _default_input({ + "dry_lab_only": False, + "safety_checks": {"toxicity_screened": True, "hemolysis_screened": True, "dual_use_reviewed": True}, + "candidates": [ + {"candidate_id": "C-001", "proof_ladder_level": 2}, + ], + }) + global_results = run_global_checks(input_data) + candidate_results = run_candidate_checks(input_data["candidates"]) + result = build_filter_result(input_data, global_results, candidate_results) + assert result["global_pass"] is False + assert result["summary"]["overall_filter_pass"] is False + assert result["summary"]["passed"] == 1 # candidates pass but global fails + + def test_markdown_contains_sections(self): + """Markdown output contains expected sections.""" + input_data = _default_input() + global_results = run_global_checks(input_data) + candidate_results = run_candidate_checks(input_data["candidates"]) + result = build_filter_result(input_data, global_results, candidate_results) + md = build_markdown_result(result) + assert "# Safe Publication Filter Result" in md + assert "## Filter Metadata" in md + assert "## Global Checks" in md + assert "## Summary" in md + assert "## Per-Candidate Results" in md + assert "## Caveats" in md + assert "CAND-001" in md + assert "CAND-002" in md + + def test_markdown_summary_counts(self): + """Markdown contains summary counts.""" + input_data = _default_input() + global_results = run_global_checks(input_data) + candidate_results = run_candidate_checks(input_data["candidates"]) + result = build_filter_result(input_data, global_results, candidate_results) + md = build_markdown_result(result) + assert "Total candidates" in md + assert "Passed" in md + assert "Failed" in md + assert "overall filter verdict" in md.lower() + + def test_load_input_valid(self): + """Loading a valid JSON file succeeds.""" + example = _REPO_DIR / "examples" / "safe_publication_filter_example_input.json" + assert example.exists() + data = load_input(str(example)) + assert "candidates" in data + assert "dry_lab_only" in data + assert data["dry_lab_only"] is True + + def test_load_input_missing_file(self): + """Loading a nonexistent file raises FileNotFoundError.""" + with pytest.raises(FileNotFoundError): + load_input("/nonexistent/path.json") + + def test_load_input_invalid_json(self): + """Loading an invalid JSON file raises ValueError.""" + import tempfile + with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f: + f.write("not json") + p = Path(f.name) + try: + with pytest.raises(ValueError, match="Invalid JSON"): + load_input(p) + finally: + p.unlink() + + def test_cli_writes_json(self, tmp_path): + """CLI produces a valid JSON output file.""" + example = _REPO_DIR / "examples" / "safe_publication_filter_example_input.json" + out_json = tmp_path / "filter_result.json" + r = subprocess.run( + [PYTHON, "scripts/safe_publication_filter.py", + "--input", str(example), + "--out-json", str(out_json)], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert r.returncode in (0, 1) + assert out_json.exists() + data = json.loads(out_json.read_text()) + assert data["filter_metadata"]["filter_type"] == "safe_publication_filter" + assert "candidates" in data + assert "global_checks" in data + + def test_cli_exit_code_0_all_pass(self, tmp_path): + """CLI returns 0 when all checks pass.""" + input_data = _default_input() + input_file = tmp_path / "input.json" + input_file.write_text(json.dumps(input_data)) + r = subprocess.run( + [PYTHON, "scripts/safe_publication_filter.py", + "--input", str(input_file)], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert r.returncode == 0 + result = json.loads(r.stdout) + assert result["summary"]["overall_filter_pass"] is True + + def test_cli_exit_code_1_global_fail(self, tmp_path): + """CLI returns 1 when global checks fail.""" + input_data = _default_input({"dry_lab_only": False}) + input_file = tmp_path / "input.json" + input_file.write_text(json.dumps(input_data)) + r = subprocess.run( + [PYTHON, "scripts/safe_publication_filter.py", + "--input", str(input_file)], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert r.returncode == 1 + result = json.loads(r.stdout) + assert result["global_pass"] is False + + def test_cli_exit_code_1_candidate_fail(self, tmp_path): + """CLI returns 1 when a candidate fails checks.""" + input_data = _default_input({ + "candidates": [{"candidate_id": "C-001", "proof_ladder_level": 6}], + }) + input_file = tmp_path / "input.json" + input_file.write_text(json.dumps(input_data)) + r = subprocess.run( + [PYTHON, "scripts/safe_publication_filter.py", + "--input", str(input_file)], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert r.returncode == 1 + result = json.loads(r.stdout) + assert result["summary"]["overall_filter_pass"] is False + + def test_cli_missing_input_fails(self, tmp_path): + """CLI returns 2 for missing input file.""" + r = subprocess.run( + [PYTHON, "scripts/safe_publication_filter.py", + "--input", str(tmp_path / "nonexistent.json")], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert r.returncode == 2 + + def test_cli_empty_candidates_fails(self, tmp_path): + """CLI returns 2 for empty candidates list.""" + input_data = _default_input({"candidates": []}) + input_file = tmp_path / "empty.json" + input_file.write_text(json.dumps(input_data)) + r = subprocess.run( + [PYTHON, "scripts/safe_publication_filter.py", + "--input", str(input_file)], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert r.returncode == 2 + + def test_cli_produces_json_and_md(self, tmp_path): + """CLI produces both JSON and Markdown output when requested.""" + example = _REPO_DIR / "examples" / "safe_publication_filter_example_input.json" + out_json = tmp_path / "result.json" + out_md = tmp_path / "result.md" + r = subprocess.run( + [PYTHON, "scripts/safe_publication_filter.py", + "--input", str(example), + "--out-json", str(out_json), + "--out-md", str(out_md)], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert r.returncode in (0, 1) + assert out_json.exists() + assert out_md.exists() + md_content = out_md.read_text() + assert "# Safe Publication Filter Result" in md_content + + def test_filter_result_has_caveats(self): + """Filter result always includes caveats.""" + input_data = _default_input() + global_results = run_global_checks(input_data) + candidate_results = run_candidate_checks(input_data["candidates"]) + result = build_filter_result(input_data, global_results, candidate_results) + assert len(result["caveats"]) >= 6 + assert any("not biological proof" in c for c in result["caveats"]) + assert result["dry_lab_only"] is True + + def test_filter_metadata_structure(self): + """Filter metadata has correct structure.""" + input_data = _default_input() + global_results = run_global_checks(input_data) + candidate_results = run_candidate_checks(input_data["candidates"]) + result = build_filter_result(input_data, global_results, candidate_results) + meta = result["filter_metadata"] + assert meta["filter_type"] == "safe_publication_filter" + assert meta["filter_version"] == "1.0.0" + assert meta["input_source"] == "test-batch" + assert meta["pipeline_version"] == "v0.5.75-test" + assert "generated_at" in meta + + def test_example_input_produces_valid_result(self): + """The example input file produces a valid filter result with expected outcomes.""" + example = _REPO_DIR / "examples" / "safe_publication_filter_example_input.json" + data = load_input(str(example)) + global_results = run_global_checks(data) + candidate_results = run_candidate_checks(data["candidates"]) + result = build_filter_result(data, global_results, candidate_results) + assert result["global_pass"] is True + s = result["summary"] + assert s["total_candidates"] == 4 + outcomes = {r["overall"] for r in result["candidates"]} + # CAND-003 has pll=5 → FAIL + # CAND-002 has pll=4 → WARN + assert "FAIL" in outcomes + assert "WARN" in outcomes + assert "PASS" in outcomes + + def test_example_input_cli_round_trip(self, tmp_path): + """Running the CLI on the example input produces a parseable result.""" + example = _REPO_DIR / "examples" / "safe_publication_filter_example_input.json" + out_json = tmp_path / "result.json" + r = subprocess.run( + [PYTHON, "scripts/safe_publication_filter.py", + "--input", str(example), + "--out-json", str(out_json)], + capture_output=True, text=True, + env={"PYTHONPATH": "src"}, + ) + assert r.returncode == 1 # CAND-003 fails + assert out_json.exists() + data = json.loads(out_json.read_text()) + assert data["summary"]["failed"] == 1 + assert data["summary"]["warned"] == 1 + assert data["summary"]["passed"] == 2 diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index 09cbe83f..2a0a8718 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -2,9 +2,9 @@ import subprocess import sys -# Baseline: 2724 tests as of Loop 74 (2026-07-09) — Phase F F4 (failed-candidate report generator) +# Baseline: 2758 tests as of Loop 75 (2026-07-09) — Phase F F5 (safe-publication filter) # Raise threshold if legitimate tests are removed; lower threshold is okay. -BASELINE = 2724 +BASELINE = 2758 TOLERANCE = 0.05 # allow 5% fluctuation