Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/research/NEXT_100_PR_MAP.md
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ Make qualified external review easier and safer.
| E1 | Add external review packet schema (complete). — evidence/external_review_packet.py: ERP- schema with 14 fields, validation rules, dry_lab_only enforcement; tests/evidence/test_external_review_packet.py + test_external_review_packet_schema.py. | Packets become machine-checkable. | C/D |
| E2 | Add example external review packet using toy data (complete). — evidence/example_external_review_packet.py: toy-data example ERP with MOCK-/TOY- prefix enforcement and is_example_data=True; tests/evidence/test_example_external_review_packet.py. | Partners know what to expect. | C/D |
| E3 | Add reviewer questionnaire schema (complete). | Makes external review feedback machine-readable: Likert clarity ratings for activity/safety/novelty claims, synthesis recommendation, structured comments. | B/C |
| E4 | Add safety-release decision schema. | Release review becomes auditable. | D |
| E4 | Add safety-release decision schema. | Release review becomes auditable. | D | DONE |
| E5 | Add non-protocol pilot pre-registration schema (complete). — evidence/pilot_preregistration.py: non-protocol pilot pre-registration schema (PPR-) freezing selection logic before batch release; tests/evidence/test_pilot_preregistration.py + test_pilot_preregistration_schema.py. | Freezes selection logic. | C/D |
| E6 | Add packet generator CLI (complete). — scripts/generate_review_packet.py: generates skeleton external review packet JSON; make generate-review-packet target; validates against schemas/external_review_packet.schema.json; dry_lab_only_attestation=True enforced. | Reduces manual packaging errors. | C/D |
| E7 | Add packet validator CLI (complete). — src/openamp_foundry/cli/commands/validate_packet.py: load_packet_from_json() reads ERP- JSON from disk; validate_packet_file() returns {valid, violations, packet_id, error}; _run_validate_packet() prints PASS/FAIL with violations; 45 tests in tests/cli/test_validate_packet.py. | Review readiness becomes testable. | C/D |
Expand Down
165 changes: 165 additions & 0 deletions src/openamp_foundry/evidence/safety_release_decision.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
"""SRD- safety-release decision schema.

Machine-checkable record of whether a candidate package was authorized for
external release, which safety checks were passed, what restrictions apply,
and why it was rejected if not authorized.

Makes release review auditable: every external release of pipeline outputs
must reference an SRD- record. No release is implicit or unrecorded.
"""

from __future__ import annotations

from dataclasses import dataclass

VALID_SRD_DECISIONS: frozenset[str] = frozenset({
"authorized",
"rejected",
"pending_review",
})

VALID_RELEASE_SCOPES: frozenset[str] = frozenset({
"academic_collaboration",
"public_preprint",
"internal_only",
"restricted_partner",
})

VALID_SAFETY_CHECK_IDS: frozenset[str] = frozenset({
"dual_use_screened",
"toxicity_flags_reviewed",
"hemolysis_flags_reviewed",
"novelty_claims_bounded",
"dry_lab_only_label_present",
"no_sequence_privacy_violation",
"candidate_count_within_limit",
})

REQUIRED_SAFETY_CHECKS: tuple[str, ...] = (
"dual_use_screened",
"dry_lab_only_label_present",
"novelty_claims_bounded",
)


@dataclass
class SafetyReleaseDecision:
srd_id: str
pipeline_version: str
erp_id: str
release_decision: str
release_scope: str
safety_checks_passed: list[str]
restrictions: list[str]
rejection_reason: str
all_required_checks_passed: bool
dry_lab_only: bool
limitations: list[str]
created_at: str


def validate_safety_release_decision(srd: SafetyReleaseDecision) -> None:
if not srd.srd_id.startswith("SRD-"):
raise ValueError(f"srd_id must start with 'SRD-': {srd.srd_id!r}")
if not srd.pipeline_version:
raise ValueError("pipeline_version must be non-empty")
if not srd.erp_id.startswith("ERP-"):
raise ValueError(f"erp_id must start with 'ERP-': {srd.erp_id!r}")
if srd.release_decision not in VALID_SRD_DECISIONS:
raise ValueError(
f"release_decision {srd.release_decision!r} not in VALID_SRD_DECISIONS"
)
if srd.release_scope not in VALID_RELEASE_SCOPES:
raise ValueError(
f"release_scope {srd.release_scope!r} not in VALID_RELEASE_SCOPES"
)
for check_id in srd.safety_checks_passed:
if check_id not in VALID_SAFETY_CHECK_IDS:
raise ValueError(
f"safety check {check_id!r} not in VALID_SAFETY_CHECK_IDS"
)
required_passed = all(
req in srd.safety_checks_passed for req in REQUIRED_SAFETY_CHECKS
)
if srd.all_required_checks_passed != required_passed:
raise ValueError(
"all_required_checks_passed mismatch with safety_checks_passed"
)
if srd.release_decision == "authorized" and not srd.all_required_checks_passed:
raise ValueError(
"release_decision='authorized' requires all_required_checks_passed=True"
)
if srd.release_decision == "rejected" and not srd.rejection_reason:
raise ValueError(
"rejection_reason must be non-empty when release_decision='rejected'"
)
if not srd.dry_lab_only:
raise ValueError("dry_lab_only must be True")
if not srd.limitations:
raise ValueError("limitations must be non-empty")
if not srd.created_at:
raise ValueError("created_at must be non-empty")


def build_safety_release_decision(
*,
srd_id: str,
pipeline_version: str,
erp_id: str,
release_decision: str,
release_scope: str,
safety_checks_passed: list[str],
restrictions: list[str],
rejection_reason: str = "",
limitations: list[str],
created_at: str,
) -> SafetyReleaseDecision:
"""Build a SafetyReleaseDecision.

safety_checks_passed: list of VALID_SAFETY_CHECK_IDS that were verified.
all_required_checks_passed is auto-computed from REQUIRED_SAFETY_CHECKS.
"""
required_passed = all(
req in safety_checks_passed for req in REQUIRED_SAFETY_CHECKS
)
srd = SafetyReleaseDecision(
srd_id=srd_id,
pipeline_version=pipeline_version,
erp_id=erp_id,
release_decision=release_decision,
release_scope=release_scope,
safety_checks_passed=list(safety_checks_passed),
restrictions=list(restrictions),
rejection_reason=rejection_reason,
all_required_checks_passed=required_passed,
dry_lab_only=True,
limitations=limitations,
created_at=created_at,
)
validate_safety_release_decision(srd)
return srd


def format_safety_release_decision(srd: SafetyReleaseDecision) -> str:
lines = [
f"Safety-Release Decision — {srd.srd_id}",
f"Pipeline: {srd.pipeline_version} | ERP: {srd.erp_id}",
f"Decision: {srd.release_decision} | Scope: {srd.release_scope}",
f"All required checks passed: {srd.all_required_checks_passed}",
]
lines.append(f"Safety checks passed ({len(srd.safety_checks_passed)}):")
for check in srd.safety_checks_passed:
lines.append(f" + {check}")
missing = [r for r in REQUIRED_SAFETY_CHECKS if r not in srd.safety_checks_passed]
if missing:
lines.append("Required checks NOT passed:")
for m in missing:
lines.append(f" ! {m}")
if srd.restrictions:
lines.append(f"Restrictions: {'; '.join(srd.restrictions)}")
if srd.rejection_reason:
lines.append(f"Rejection reason: {srd.rejection_reason}")
lines.append(f"Created: {srd.created_at}")
lines.append(f"Limitations: {'; '.join(srd.limitations)}")
lines.append(f"dry_lab_only: {srd.dry_lab_only}")
return "\n".join(lines)
Loading
Loading