|
| 1 | +from __future__ import annotations |
| 2 | +from dataclasses import dataclass, field |
| 3 | +from typing import List, Optional |
| 4 | + |
| 5 | + |
| 6 | +VALID_SELECTION_BASES = frozenset({ |
| 7 | + "score_rank", # selected primarily by pipeline score rank |
| 8 | + "safety_gate_passed", # selected because it passed all safety screens |
| 9 | + "novelty_evidence", # selected due to novelty evidence (CFC/FNR) |
| 10 | + "benchmark_performance", # selected due to benchmark performance vs baselines |
| 11 | + "expert_review", # selected following expert human review |
| 12 | + "calibration_signal", # selected based on calibration-adjusted confidence |
| 13 | + "diversity_criteria", # selected to maximize batch diversity |
| 14 | + "not_specified", # selection basis not documented (should be avoided) |
| 15 | +}) |
| 16 | + |
| 17 | +VALID_LEDGER_COMPLETENESS_GRADES = frozenset({ |
| 18 | + "complete", # all expected artifact references are present |
| 19 | + "partial", # some expected artifacts are missing but core selection documented |
| 20 | + "incomplete", # major artifacts missing; selection not fully traceable |
| 21 | + "not_assessed", # completeness not checked |
| 22 | +}) |
| 23 | + |
| 24 | +REQUIRED_SELECTION_ARTIFACT_TYPES = frozenset({ |
| 25 | + "BSP", # batch selection proposal |
| 26 | + "SEG", # score vs enemy gap report |
| 27 | + "SRG", # scientific review readiness gate |
| 28 | +}) |
| 29 | + |
| 30 | +VALID_SELECTION_OUTCOME_TYPES = frozenset({ |
| 31 | + "selected_for_pilot", |
| 32 | + "selected_for_next_batch", |
| 33 | + "selected_for_external_review", |
| 34 | + "deselected_safety", |
| 35 | + "deselected_low_score", |
| 36 | + "deselected_insufficient_evidence", |
| 37 | + "held_pending_review", |
| 38 | +}) |
| 39 | + |
| 40 | + |
| 41 | +@dataclass |
| 42 | +class ArtifactReference: |
| 43 | + artifact_type: str |
| 44 | + artifact_id: str |
| 45 | + influence_description: str |
| 46 | + |
| 47 | + |
| 48 | +@dataclass |
| 49 | +class SelectionTransparencyLedger: |
| 50 | + stl_id: str |
| 51 | + run_id: str |
| 52 | + candidate_family_id: str |
| 53 | + selection_basis: str |
| 54 | + selection_outcome: str |
| 55 | + completeness_grade: str |
| 56 | + artifact_references: List[ArtifactReference] |
| 57 | + n_artifacts_referenced: int |
| 58 | + has_required_artifacts: bool |
| 59 | + missing_required_artifact_types: List[str] |
| 60 | + seg_id: Optional[str] |
| 61 | + srg_id: Optional[str] |
| 62 | + bsp_id: Optional[str] |
| 63 | + dry_lab_only: bool |
| 64 | + limitations: str |
| 65 | + notes: str = "" |
| 66 | + |
| 67 | + |
| 68 | +@dataclass |
| 69 | +class STLValidationResult: |
| 70 | + valid: bool |
| 71 | + violations: List[str] = field(default_factory=list) |
| 72 | + |
| 73 | + |
| 74 | +def validate_selection_transparency_ledger(ledger: SelectionTransparencyLedger) -> STLValidationResult: |
| 75 | + violations = [] |
| 76 | + |
| 77 | + if not ledger.stl_id.startswith("STL-"): |
| 78 | + violations.append("stl_id must start with 'STL-'") |
| 79 | + |
| 80 | + if not ledger.run_id: |
| 81 | + violations.append("run_id is required") |
| 82 | + |
| 83 | + if not ledger.candidate_family_id: |
| 84 | + violations.append("candidate_family_id is required") |
| 85 | + |
| 86 | + if ledger.candidate_family_id.startswith("TOY-"): |
| 87 | + violations.append("TOY- candidate_family_id is not allowed in real STL- records") |
| 88 | + |
| 89 | + if ledger.selection_basis not in VALID_SELECTION_BASES: |
| 90 | + violations.append( |
| 91 | + f"selection_basis '{ledger.selection_basis}' must be one of {sorted(VALID_SELECTION_BASES)}" |
| 92 | + ) |
| 93 | + |
| 94 | + if ledger.selection_outcome not in VALID_SELECTION_OUTCOME_TYPES: |
| 95 | + violations.append( |
| 96 | + f"selection_outcome '{ledger.selection_outcome}' must be one of {sorted(VALID_SELECTION_OUTCOME_TYPES)}" |
| 97 | + ) |
| 98 | + |
| 99 | + if ledger.completeness_grade not in VALID_LEDGER_COMPLETENESS_GRADES: |
| 100 | + violations.append( |
| 101 | + f"completeness_grade '{ledger.completeness_grade}' must be one of {sorted(VALID_LEDGER_COMPLETENESS_GRADES)}" |
| 102 | + ) |
| 103 | + |
| 104 | + if ledger.n_artifacts_referenced != len(ledger.artifact_references): |
| 105 | + violations.append( |
| 106 | + f"n_artifacts_referenced ({ledger.n_artifacts_referenced}) must match len(artifact_references) ({len(ledger.artifact_references)})" |
| 107 | + ) |
| 108 | + |
| 109 | + if ledger.n_artifacts_referenced < 1: |
| 110 | + violations.append("at least one artifact_reference is required") |
| 111 | + |
| 112 | + # Validate each artifact reference |
| 113 | + for i, ref in enumerate(ledger.artifact_references): |
| 114 | + if not ref.artifact_type: |
| 115 | + violations.append(f"artifact_references[{i}].artifact_type is required") |
| 116 | + if not ref.artifact_id: |
| 117 | + violations.append(f"artifact_references[{i}].artifact_id is required") |
| 118 | + if not ref.influence_description or len(ref.influence_description.strip()) < 5: |
| 119 | + violations.append(f"artifact_references[{i}].influence_description must be at least 5 chars") |
| 120 | + |
| 121 | + # Check presence of required artifacts |
| 122 | + present_types = {ref.artifact_type for ref in ledger.artifact_references} |
| 123 | + missing = sorted(REQUIRED_SELECTION_ARTIFACT_TYPES - present_types) |
| 124 | + if sorted(ledger.missing_required_artifact_types) != missing: |
| 125 | + violations.append( |
| 126 | + f"missing_required_artifact_types {ledger.missing_required_artifact_types} does not match computed {missing}" |
| 127 | + ) |
| 128 | + |
| 129 | + expected_has_required = len(missing) == 0 |
| 130 | + if ledger.has_required_artifacts != expected_has_required: |
| 131 | + violations.append( |
| 132 | + f"has_required_artifacts ({ledger.has_required_artifacts}) inconsistent with missing types ({missing})" |
| 133 | + ) |
| 134 | + |
| 135 | + # complete grade requires has_required_artifacts |
| 136 | + if ledger.completeness_grade == "complete" and not ledger.has_required_artifacts: |
| 137 | + violations.append( |
| 138 | + "completeness_grade='complete' requires has_required_artifacts=True" |
| 139 | + ) |
| 140 | + |
| 141 | + # incomplete grade must not have all required artifacts |
| 142 | + if ledger.completeness_grade == "incomplete" and ledger.has_required_artifacts and not missing: |
| 143 | + violations.append( |
| 144 | + "completeness_grade='incomplete' should not have all required artifacts present" |
| 145 | + ) |
| 146 | + |
| 147 | + # not_specified selection_basis is a weak signal |
| 148 | + if ledger.selection_basis == "not_specified": |
| 149 | + violations.append( |
| 150 | + "selection_basis='not_specified' is not allowed; document the actual selection basis" |
| 151 | + ) |
| 152 | + |
| 153 | + # ID field prefix checks |
| 154 | + if ledger.seg_id is not None and not ledger.seg_id.startswith("SEG-"): |
| 155 | + violations.append("seg_id must start with 'SEG-' when provided") |
| 156 | + |
| 157 | + if ledger.srg_id is not None and not ledger.srg_id.startswith("SRG-"): |
| 158 | + violations.append("srg_id must start with 'SRG-' when provided") |
| 159 | + |
| 160 | + if ledger.bsp_id is not None and not ledger.bsp_id.startswith("BSP-"): |
| 161 | + violations.append("bsp_id must start with 'BSP-' when provided") |
| 162 | + |
| 163 | + if not ledger.dry_lab_only: |
| 164 | + violations.append("dry_lab_only must be True for STL- records") |
| 165 | + |
| 166 | + if not ledger.limitations or len(ledger.limitations.strip()) < 10: |
| 167 | + violations.append("limitations must be a non-empty string (at least 10 characters)") |
| 168 | + |
| 169 | + return STLValidationResult(valid=len(violations) == 0, violations=violations) |
| 170 | + |
| 171 | + |
| 172 | +def build_selection_transparency_ledger( |
| 173 | + stl_id: str, |
| 174 | + run_id: str, |
| 175 | + candidate_family_id: str, |
| 176 | + selection_basis: str, |
| 177 | + selection_outcome: str, |
| 178 | + artifact_references: List[ArtifactReference], |
| 179 | + limitations: str, |
| 180 | + notes: str = "", |
| 181 | + seg_id: Optional[str] = None, |
| 182 | + srg_id: Optional[str] = None, |
| 183 | + bsp_id: Optional[str] = None, |
| 184 | +) -> SelectionTransparencyLedger: |
| 185 | + present_types = {ref.artifact_type for ref in artifact_references} |
| 186 | + missing = sorted(REQUIRED_SELECTION_ARTIFACT_TYPES - present_types) |
| 187 | + has_required = len(missing) == 0 |
| 188 | + |
| 189 | + if has_required and len(artifact_references) >= len(REQUIRED_SELECTION_ARTIFACT_TYPES): |
| 190 | + completeness_grade = "complete" |
| 191 | + elif missing and len(missing) < len(REQUIRED_SELECTION_ARTIFACT_TYPES): |
| 192 | + completeness_grade = "partial" |
| 193 | + elif not artifact_references: |
| 194 | + completeness_grade = "incomplete" |
| 195 | + else: |
| 196 | + completeness_grade = "partial" |
| 197 | + |
| 198 | + ledger = SelectionTransparencyLedger( |
| 199 | + stl_id=stl_id, |
| 200 | + run_id=run_id, |
| 201 | + candidate_family_id=candidate_family_id, |
| 202 | + selection_basis=selection_basis, |
| 203 | + selection_outcome=selection_outcome, |
| 204 | + completeness_grade=completeness_grade, |
| 205 | + artifact_references=artifact_references, |
| 206 | + n_artifacts_referenced=len(artifact_references), |
| 207 | + has_required_artifacts=has_required, |
| 208 | + missing_required_artifact_types=missing, |
| 209 | + seg_id=seg_id, |
| 210 | + srg_id=srg_id, |
| 211 | + bsp_id=bsp_id, |
| 212 | + dry_lab_only=True, |
| 213 | + limitations=limitations, |
| 214 | + notes=notes, |
| 215 | + ) |
| 216 | + result = validate_selection_transparency_ledger(ledger) |
| 217 | + if not result.valid: |
| 218 | + raise ValueError(f"Invalid STL: {result.violations}") |
| 219 | + return ledger |
| 220 | + |
| 221 | + |
| 222 | +def format_selection_transparency_ledger(ledger: SelectionTransparencyLedger) -> str: |
| 223 | + lines = [ |
| 224 | + f"Selection Transparency Ledger — {ledger.stl_id}", |
| 225 | + f"Run: {ledger.run_id} | Family: {ledger.candidate_family_id}", |
| 226 | + f"Selection Basis: {ledger.selection_basis}", |
| 227 | + f"Outcome: {ledger.selection_outcome}", |
| 228 | + f"Completeness Grade: {ledger.completeness_grade}", |
| 229 | + f"Artifacts Referenced: {ledger.n_artifacts_referenced}", |
| 230 | + f"Required Artifacts Present: {ledger.has_required_artifacts}", |
| 231 | + ] |
| 232 | + if ledger.missing_required_artifact_types: |
| 233 | + lines.append(f"Missing Required: {', '.join(ledger.missing_required_artifact_types)}") |
| 234 | + for ref in ledger.artifact_references: |
| 235 | + lines.append(f" [{ref.artifact_type}] {ref.artifact_id}: {ref.influence_description}") |
| 236 | + if ledger.seg_id: |
| 237 | + lines.append(f"SEG Link: {ledger.seg_id}") |
| 238 | + if ledger.srg_id: |
| 239 | + lines.append(f"SRG Link: {ledger.srg_id}") |
| 240 | + if ledger.bsp_id: |
| 241 | + lines.append(f"BSP Link: {ledger.bsp_id}") |
| 242 | + lines.append(f"Limitations: {ledger.limitations}") |
| 243 | + if ledger.notes: |
| 244 | + lines.append(f"Notes: {ledger.notes}") |
| 245 | + lines.append("dry_lab_only: True") |
| 246 | + return "\n".join(lines) |
0 commit comments