diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 15cea45..8121ffb 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -31,13 +31,25 @@ src/sapiens/ budget.py cooperative step/time budgets and preemption exceptions queue.py bounded SQLite job queue with leases/idempotency daemon.py bounded background worker skeleton - cli.py synthetic-only demo + cli.py synthetic-only demo + `gates` subcommand adapters/ synthetic adapters + Phase-4 real-data Kepler adapter (published-signal re-derivation; validators sandboxed behind the adapter contract) + gates/ discovery-gate hardening (scout Phases 0–5), stdlib-only: + surprise.py G-03 literature-measured surprise + G-07 robust full-set baseline + nulls.py Phase-1 mandatory null-provenance records (+FP-04 state) + fdr.py family-wide Benjamini-Hochberg + sigma<->p conversions + thresholds.py decoupled ENTRY/RANK/CONFIRM + B-02/B-03/B-05 fixes + devils_advocate.py B-06 permutation test on the confound-independence null + promotion.py recalibrated promotion_score + anomaly_priority (G-05/G-06/FP-06) + pipeline.py per-family orchestration -> GateOutcome + bounded shortlist + corpus.py gaming vectors + ~20 historical positives + decoys + blind.py Phase-3 blind re-run harness (strip/expand/custody/grade) + dossier.py Phase-4 referee dossier, tiered authority, override log + criteria.py Phase-5 automated success-criteria suite ``` -Dependency rule: `sapiens` core modules do not import `sapiens.adapters`; tests enforce this. Adapters import core contracts. +Dependency rule: `sapiens` core modules and `sapiens.gates` do not import `sapiens.adapters`; tests enforce this. Adapters import core contracts. The `gates` subpackage is self-contained (stdlib only) and copies no ASTRA/GEODISC/BIODISC/SLATE code — a test enforces the clean-room invariant on it too. ## DomainAdapter contract @@ -78,4 +90,35 @@ Hash chaining detects tampering but does **not** prove authorship, scientific tr `WorkQueue` gives bounded jobs, serialized payload-size limits, idempotency keys, leases, stale-lease rejection, and retry/dead states. `DiscoveryDaemon.run_bounded` executes only explicitly registered handlers under time/step budgets; it does not dynamically import or shell out from queue payloads. +## Discovery-gate hardening (`sapiens.gates`, scout Phases 0–5) + +A separate, stdlib-only subpackage that hardens the *discovery-decision* gates +against gaming and null-model incompleteness. It does not touch the L0→L4 +foundation; it consumes a domain-neutral `GateInputs` (adapters translate their +measurements into it) and returns a `GateOutcome`/`FamilyResult`. + +Data flow for a candidate family: `evaluate_family` builds a family-wide +Benjamini-Hochberg p-value set from each candidate's **null-provenance** +σ (`nulls.NullProvenance`, only counted when the null is *calibratable* — data +fetched and no un-excluded instrument systematic); computes `promotion_score` +(recalibrated weights, never penalising mechanism-absence) and +`anomaly_priority` (mechanism-absence boost gated by **literature-measured +surprise**, G-03, and the **conservation-law guard**, FP-06); then applies the +decoupled thresholds — **ENTRY** = 3σ-equiv AND family-FDR survival AND a +calibratable null (so an UNCALIBRATED candidate is *surfaced, never admitted*, +G-06); **RANK** = continuous `promotion + anomaly`; **CONFIRM** = domain-specific +claim bar. Reserved paradigm-breaker slots (`reserved_slot_eligible`, G-05: +`promotion_score ≥ 0.30` + surprise + the UNEXPLAINED_CONFIRMED signature) get +guaranteed representation in a bounded top-K shortlist. + +The G-07 baseline receipt (`surprise.robust_baseline`) computes median/MAD-σ over +**every** point and records the count, so a "clean-subset" gamed baseline is +structurally impossible through the API. `criteria.run_success_criteria` runs the +five Phase-5 ship gates; `blind` provides the sealed, separate-custody re-run +harness; `dossier` builds the human referee bundle (with the strongest +disconfirming explanation forced next to each claim) and enforces that only +CONFIRM-tier candidates are autonomous-claim-eligible. Everything is a pure +function or frozen dataclass; nothing here claims a discovery +(`scientific_discoveries_claimed = 0` remains true by construction). + Synthetic and CORE (first-party, real-data) adapters run in-process with cooperative preemption. UNTRUSTED (third-party) adapters run only via `sapiens.isolation`: a child process applies POSIX rlimits (CPU, address space, open files) to itself, the parent enforces a wall-clock timeout, and every failure mode is contained fail-closed (no evidence on failure). Third-party adapters additionally require a recorded owner permission (`permissions.json`, empty by default) before the registry will validate them at all. rlimits bound resources; they are not a full security sandbox. diff --git a/README.md b/README.md index da074ef..9db2f50 100644 --- a/README.md +++ b/README.md @@ -12,9 +12,12 @@ SAPIENS is an **experimental platform for traceable cross-domain scientific-discovery workflows** — a `DomainAdapter` boundary, a hash-chained -L0→L4 evidence ledger, and kernel-gated promotions. **Phases 0–4 shipped (v0.5.0): three -synthetic adapters plus one real-data Kepler photometry adapter that -re-derives a published signal — no discoveries claimed.** +L0→L4 evidence ledger, and kernel-gated promotions. **Foundation phases 0–4 +shipped: three synthetic adapters plus one real-data Kepler photometry adapter +that re-derives a published signal. v0.6.0 adds the `sapiens.gates` +discovery-gate-hardening track (anti-gaming seams, a mandatory logged null +layer, decoupled thresholds, a blind re-run harness, and a human-gate dossier +layer). No discoveries claimed.** It provides the plumbing a discovery system needs before it can be trusted: a domain-neutral adapter boundary, an append-only hash-chained evidence @@ -68,11 +71,17 @@ src/sapiens/ discovery.py autonomous discovery driver — proposes candidates and climbs them L0→L3 under budgets via the daemon; L4 stays human-gated - cli.py synthetic-only demo entrypoint + cli.py synthetic-only demo entrypoint (+ `gates` subcommand) adapters/ three deterministic synthetic adapters (linear, threshold, periodic-signal photometry) plus the Phase-4 real-data Kepler photometry adapter (re-derivation of a published signal; no discovery claimed) + gates/ discovery-gate hardening (scout Phases 0–5), stdlib-only: + anti-gaming seams (G-03/05/06/07 + FP-06 guard), a mandatory + logged null layer + family-wide BH-FDR, decoupled + ENTRY/RANK/CONFIRM thresholds, a blind re-run harness, a + human-gate dossier layer, and an automated success-criteria + suite — self-contained, imports no adapters, claims nothing ``` Key design rules (enforced by tests in [`tests/`](tests/)): @@ -97,6 +106,12 @@ Key design rules (enforced by tests in [`tests/`](tests/)): every time. - **Bounded autonomy** — background work runs through a bounded SQLite queue with leases, cooperative budgets, and preemption; no unbounded loops. +- **Gate-hardening is clean-room too** — `sapiens.gates` implements the + discovery-decision hardening (anti-gaming + null layer + decoupled thresholds + + blind harness + human-gate dossier) from spec only; a test enforces it + imports no adapters and copies no ASTRA-family code. Run it with + `sapiens gates` (full success-criteria suite) or `sapiens gates --mode gaming` + (the Phase-0 gaming re-test, which reports **0 leaks**). Full details: [`ARCHITECTURE.md`](ARCHITECTURE.md) · [`VALIDATION.md`](VALIDATION.md) · [`PROVENANCE.md`](PROVENANCE.md) · diff --git a/ROADMAP.md b/ROADMAP.md index e3debfb..e66c332 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -102,3 +102,84 @@ Real domain adapters: - Human L4 gates. - Reproduction bundles. - Prediction tracking and demotion on contradicting evidence. + +--- + +# Discovery-Gate Hardening (scout HOW-TO-PROCEED Phases 0–5) — shipped v0.6.0 + +> **Distinct track, distinct numbering.** The *foundation* phases above (0–4) +> built the ledger/adapter/review plumbing. This second track hardens the +> **discovery-decision gates** against gaming and null-model incompleteness, +> following `beast-scout`'s blind red-team benchmark. It ships as the +> self-contained, stdlib-only `sapiens.gates` subpackage and changes **no** +> foundation behaviour. Honesty invariant unchanged: +> `scientific_discoveries_claimed = 0`. + +Status legend: **code-complete** (implemented + tested) · **library-complete** +(decision-support code done; the surrounding human workflow is process, not +code) · **synthetic-corpus** (validated on spec-derived synthetic profiles, not +real datasets — the real measurement is a prospective trial). + +### DG-0 — weld the 4 gaming seams — **code-complete** +- **G-03** anomaly boost is conditioned on *literature-measured surprise* + (`gates.surprise`), never mere absence of a mechanism. +- **G-05** reserved paradigm-breaker slots require `promotion_score >= 0.30` + **and** the G-03 surprise condition (`gates.promotion.reserved_slot_eligible`). +- **G-06** L2-holdout-passed is a *prerequisite* for CALIBRATED; the additive + score only ranks *within* the calibrated set (`gates.promotion`, + `gates.pipeline` entry gate). +- **G-07** MAD/σ baseline is computed on the **full dataset** with a robust + estimator that consumes every point (`gates.surprise.robust_baseline`). +- **FP-06** conservation-law guard: a violation ⇒ correct null is measurement + error ⇒ no boost/entry without reproducible orthogonal confirmation. +- **Exit met:** the 7 gaming vectors re-run to **0 leaks** (`gates.criteria.run_gaming_retest`; test `test_phase0_exit_zero_gaming_leaks`). + +### DG-1 — mandatory, logged null layer — **code-complete** +- Every candidate carries a `NullProvenance` record (which null, external data + required/fetched y/n, σ-under-null); no null ⇒ **UNCALIBRATED**, never a + silent pass (`gates.nulls`). +- **Family-wide** BH-FDR across the candidate family (`gates.fdr`) — the + multiplicity/look-elsewhere control (FP-09/FP-10). +- **FP-04** "instrument-systematic-not-excluded" is an explicit state that + surfaces to the human instead of passing as a clean 6σ. + +### DG-2 — decoupled threshold architecture — **code-complete** +- **ENTRY** = 3σ-equiv AND survives family BH-FDR q<0.05; **RANK** = continuous + σ/effect; **CONFIRM** = 5σ (physics) / FDR<0.05 + orthogonal replication (bio) + / formal proof-check (math) (`gates.thresholds`, `gates.pipeline`). +- Boundary fixes: **B-02** adaptive `ci_floor`, **B-03** hash-committed + thresholds, **B-05** continuous degree-of-calibration, **B-06** DevilsAdvocate + r≥0.90 → permutation test at ~3σ-equiv (`gates.devils_advocate`). + +### DG-3 — genuine blind re-run harness — **code-complete (synthetic-corpus)** +- Strip `ground_truth`/`expected_verdict`/`gates_probed`/`how_*`; expand with + ~20 historical positives + fresh decoys; separate-custody key with a + published SHA-256 commitment; blind scoring runner + grade + (`gates.blind`, `gates.corpus`). +- *Honest scope:* the positives/decoys are **spec-derived synthetic profiles**, + and separate custody is implemented in-process. A real sealed set on real + post-cutoff data with a genuine second-party custody ceremony is the + next-stage (external) test this unblocks. + +### DG-4 — human-in-loop final gate — **library-complete** +- Per-candidate dossier (σ, FDR-q, null used, external-data y/n, replication + status, UNCALIBRATED/instrument flags, and the single strongest + **disconfirming** explanation next to the claim); tiered authority + (only CONFIRM-tier is autonomous-claim-eligible, human co-sign still required); + bounded top-K throttle; override log feeding recalibration (`gates.dossier`, + `gates.pipeline` shortlist). +- *Honest scope:* this is the decision-support **data + eligibility layer**; the + reviewer UI/workflow itself is out of repo scope. + +### DG-5 — success-criteria suite — **code-complete (synthetic-corpus)** +- Automated `gates.criteria.run_success_criteria` checks all five ship gates: + 0 gaming leaks · null layer mandatory+logged (100% of shortlisted) · blind + kill-rate ≥ 8/10 + recovery ≥ 8/10 + abstention < 10% · boundary stability + ≤ 1 flip/item over 3 seeds · thresholds hash-committed per run. All pass on + the synthetic corpus (`test_success_criteria_all_pass`). +- *Honest scope:* retrospective/synthetic gates are an intuition pump. The + **only** real validation is a prospective, pre-registered precision@K / + false-positive-rate trial on a post-cutoff stream — stated in the suite output + and not faked here. + +CLI: `sapiens gates` (full suite) · `sapiens gates --mode gaming` (Phase-0 re-test). diff --git a/pyproject.toml b/pyproject.toml index 579d273..fb80121 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "sapiens-discovery" -version = "0.5.0" +version = "0.6.0" description = "Experimental foundation for traceable cross-domain scientific-discovery workflows" readme = "README.md" requires-python = ">=3.10" diff --git a/src/sapiens/__init__.py b/src/sapiens/__init__.py index 5381af4..241af04 100644 --- a/src/sapiens/__init__.py +++ b/src/sapiens/__init__.py @@ -1,5 +1,6 @@ """SAPIENS: experimental, traceable scientific-discovery workflow plumbing.""" +from . import gates from .adapter import DomainAdapter from .bridge import TransferEnvelope, transfer from .calibration import CalibrationReport, run_calibration @@ -43,9 +44,10 @@ "ValidationGates", "aggregate_confidence", "fixture_suite", + "gates", "reference_panel", "run_calibration", "score_panel", "transfer", ] -__version__ = "0.5.0" +__version__ = "0.6.0" diff --git a/src/sapiens/cli.py b/src/sapiens/cli.py index 8e007fe..cb5a5bd 100644 --- a/src/sapiens/cli.py +++ b/src/sapiens/cli.py @@ -109,6 +109,27 @@ def run_discovery( } +def run_gates(*, mode: str = "criteria") -> dict[str, object]: + """Run the discovery-gate-hardening checks (scout Phases 0-5). + + ``mode="criteria"`` runs the full Phase-5 success-criteria suite; + ``mode="gaming"`` runs only the Phase-0 gaming re-test. Neither claims a + discovery — this is evidence-hygiene plumbing. + """ + from .gates.criteria import run_gaming_retest, run_success_criteria + + if mode == "gaming": + retest = run_gaming_retest() + return { + "experimental": True, + "scientific_discoveries_claimed": 0, + "gaming_vectors_checked": retest.checked, + "gaming_leaks": list(retest.leaks), + "clean": retest.clean, + } + return run_success_criteria().to_dict() + + @contextmanager def _workdir(path: Path | None): if path is not None: @@ -134,9 +155,18 @@ def main() -> None: discover_parser.add_argument("--steps-per-job", type=int, default=20) discover_parser.add_argument("--limit-per-adapter", type=int, default=2) + gates_parser = sub.add_parser( + "gates", help="run discovery-gate-hardening checks (scout Phases 0-5)" + ) + gates_parser.add_argument( + "--mode", choices=["criteria", "gaming"], default="criteria" + ) + args = parser.parse_args() - if args.command == "discover": + if args.command == "gates": + result = run_gates(mode=args.mode) + elif args.command == "discover": with _workdir(args.workdir) as workdir: result = run_discovery( workdir, diff --git a/src/sapiens/gates/__init__.py b/src/sapiens/gates/__init__.py new file mode 100644 index 0000000..67fe508 --- /dev/null +++ b/src/sapiens/gates/__init__.py @@ -0,0 +1,129 @@ +"""Discovery-gate hardening (scout HOW-TO-PROCEED Phases 0-5). + +A self-contained, stdlib-only subpackage that implements the anti-gaming + +null-layer + decoupled-threshold + blind-harness + human-gate + success-criteria +machinery on top of the SAPIENS foundation. It imports nothing from +``sapiens.adapters`` and copies no ASTRA/GEODISC/BIODISC/SLATE code — it is built +from the recalibration/how-to-proceed specs only. No scientific discoveries are +claimed; this is evidence-hygiene plumbing. +""" + +from __future__ import annotations + +from .blind import ( + BlindReport, + CustodyKey, + SealedSet, + Verdict, + blind_run, + grade, + seal, + strip_to_blind, +) +from .corpus import ( + CorpusItem, + decoys, + full_corpus, + gaming_vectors, + historical_positives, +) +from .criteria import ( + CriterionResult, + SuiteResult, + run_gaming_retest, + run_success_criteria, +) +from .devils_advocate import PermutationResult, devils_advocate_permutation, pearson_r +from .dossier import ( + Dossier, + OverrideEvent, + OverrideLog, + autonomous_claim_eligible, + build_dossier, +) +from .fdr import FDRResult, benjamini_hochberg, pvalue_to_sigma, sigma_to_pvalue +from .nulls import InstrumentSystematic, NullKind, NullProvenance, no_null +from .pipeline import FamilyResult, GateOutcome, evaluate_family +from .promotion import ( + CalibrationStatus, + GateInputs, + LedgerStatus, + anomaly_priority, + calibration_status, + ledger_status, + promotion_score, + reserved_slot_eligible, +) +from .surprise import ( + LiteratureExpectation, + RobustBaseline, + is_surprising, + measured_surprise, + robust_baseline, +) +from .thresholds import ( + Domain, + ThresholdPolicy, + Tier, + adaptive_ci_floor, + confirm_decision, + degree_of_calibration, +) + +__all__ = [ + "BlindReport", + "CalibrationStatus", + "confirm_decision", + "CorpusItem", + "CriterionResult", + "CustodyKey", + "Domain", + "Dossier", + "FDRResult", + "FamilyResult", + "GateInputs", + "GateOutcome", + "InstrumentSystematic", + "LedgerStatus", + "LiteratureExpectation", + "NullKind", + "NullProvenance", + "OverrideEvent", + "OverrideLog", + "PermutationResult", + "RobustBaseline", + "SealedSet", + "SuiteResult", + "ThresholdPolicy", + "Tier", + "Verdict", + "adaptive_ci_floor", + "anomaly_priority", + "autonomous_claim_eligible", + "benjamini_hochberg", + "blind_run", + "build_dossier", + "calibration_status", + "decoys", + "degree_of_calibration", + "devils_advocate_permutation", + "evaluate_family", + "full_corpus", + "gaming_vectors", + "grade", + "historical_positives", + "is_surprising", + "ledger_status", + "measured_surprise", + "no_null", + "pearson_r", + "promotion_score", + "pvalue_to_sigma", + "reserved_slot_eligible", + "robust_baseline", + "run_gaming_retest", + "run_success_criteria", + "seal", + "sigma_to_pvalue", + "strip_to_blind", +] diff --git a/src/sapiens/gates/blind.py b/src/sapiens/gates/blind.py new file mode 100644 index 0000000..388fbd6 --- /dev/null +++ b/src/sapiens/gates/blind.py @@ -0,0 +1,193 @@ +"""Phase 3 — genuine blind re-run harness. + +The shipped benchmark set leaked ``ground_truth`` / ``expected_verdict`` / +``gates_probed`` / ``how_*`` **inline** — it was never sealed, so it cannot +certify the pipeline. This module provides the missing machinery: + +* **strip** the key fields from every candidate (:func:`strip_to_blind`); +* **expand** the set with historical positives + fresh decoys (from + :mod:`.corpus`); +* **separate custody**: the key is split out into a :class:`CustodyKey` whose + SHA-256 commitment is published while the labels stay sealed — the scorer runs + against the stripped inputs and never sees the key; +* **blind scoring runner** (:func:`blind_run`) executes the Phase-0/1/2-patched + pipeline on the stripped inputs and emits verdicts; +* **grade** (:func:`grade`) is the *separate-custody* step that finally opens the + key and measures kill-rate / recovery / abstention. + +The single thing this validates: does the mandatory null layer actually catch +FP-01/02/04/08 when the answer is not inline? +""" + +from __future__ import annotations + +import hashlib +import json +import random +from dataclasses import dataclass, field +from enum import Enum + +from .corpus import CorpusItem +from .pipeline import GateOutcome, evaluate_family +from .promotion import GateInputs +from .thresholds import ThresholdPolicy, Tier + + +class Verdict(str, Enum): + CLAIM = "claim" # CONFIRM tier: claim-eligible (human co-sign still required) + ENTERED = "entered" # ENTRY tier: on the shortlist as a clean signal + FLAG = "flag" # surfaced for human attention but not a clean signal + ABSTAIN = "abstain" # produced no actionable signal + + +def strip_to_blind(item: CorpusItem) -> GateInputs: + """Return only the model-visible inputs; drop every key/answer field. + + ``ground_truth`` / ``expected_verdict`` / ``gates_probed`` / + ``how_old_gate_would_kill`` / ``illegitimate_reward`` never cross into the + scored object — the whole point of Phase 3. + """ + return item.inputs + + +@dataclass(frozen=True) +class CustodyKey: + """The sealed answer key, held by a second party (separate custody).""" + + labels: dict[str, str] # candidate_id -> "positive" | "negative" + commitment: str # sha256 over the sorted labels; publishable before scoring + + @staticmethod + def build(items: list[CorpusItem]) -> CustodyKey: + labels = {it.inputs.candidate_id: it.label for it in items} + payload = json.dumps(labels, sort_keys=True, separators=(",", ":")) + return CustodyKey( + labels=labels, + commitment=hashlib.sha256(payload.encode("utf-8")).hexdigest(), + ) + + def verify_commitment(self) -> bool: + payload = json.dumps(self.labels, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() == self.commitment + + +@dataclass(frozen=True) +class SealedSet: + """A blinded candidate set + a separately-held custody key + its commitment.""" + + blind_inputs: tuple[GateInputs, ...] # the scorer sees ONLY these + key: CustodyKey # held by a second party; opened only at grade() + seed: int + + +def seal(items: list[CorpusItem], *, seed: int = 0) -> SealedSet: + """Strip, shuffle, and split custody. The scorer receives ``blind_inputs`` + and the published ``key.commitment``; the labels stay in :class:`CustodyKey`. + """ + rng = random.Random(seed) + shuffled = list(items) + rng.shuffle(shuffled) + blind = tuple(strip_to_blind(it) for it in shuffled) + return SealedSet(blind_inputs=blind, key=CustodyKey.build(items), seed=seed) + + +def outcome_to_verdict(o: GateOutcome) -> Verdict: + """Map a gate outcome onto the blind-scoring verdict vocabulary.""" + if o.tier == Tier.CONFIRM: + return Verdict.CLAIM + if o.tier == Tier.ENTRY and o.entered: + return Verdict.ENTERED + # Surfaced-but-not-a-clean-signal: reserved slot, instrument-systematic, + # or an UNCALIBRATED candidate that still carries a null sigma / surprise. + if ( + o.reserved_slot + or o.instrument_systematic_flag + or (o.sigma_under_null is not None) + or o.surprise_sigma > 0.0 + ): + return Verdict.FLAG + return Verdict.ABSTAIN + + +def blind_run( + sealed: SealedSet, + *, + policy: ThresholdPolicy | None = None, + run_id: str = "blind-run", +) -> dict[str, Verdict]: + """Run the patched pipeline on the stripped inputs — no key in scope.""" + result = evaluate_family(list(sealed.blind_inputs), policy=policy, run_id=run_id) + return {o.candidate_id: outcome_to_verdict(o) for o in result.outcomes} + + +@dataclass(frozen=True) +class BlindReport: + """Kill-rate / recovery / abstention, opened by the custody holder.""" + + positives: int + negatives: int + recovered: int # positives surfaced (CLAIM/ENTERED/FLAG) + killed: int # negatives not admitted as a clean signal (FLAG/ABSTAIN) + abstained_positives: int # positives that produced no signal (silence) + leaks: tuple[str, ...] = field(default_factory=tuple) # negatives that leaked + + @property + def recovery_rate(self) -> float: + return self.recovered / self.positives if self.positives else 0.0 + + @property + def kill_rate(self) -> float: + return self.killed / self.negatives if self.negatives else 0.0 + + @property + def abstention_rate(self) -> float: + return self.abstained_positives / self.positives if self.positives else 0.0 + + def to_dict(self) -> dict[str, object]: + return { + "positives": self.positives, + "negatives": self.negatives, + "recovered": self.recovered, + "killed": self.killed, + "recovery_rate": round(self.recovery_rate, 4), + "kill_rate": round(self.kill_rate, 4), + "abstention_rate": round(self.abstention_rate, 4), + "leaks": list(self.leaks), + } + + +def grade(verdicts: dict[str, Verdict], key: CustodyKey) -> BlindReport: + """Separate-custody grading: open the key and measure the three rates. + + * A **positive** is *recovered* if it was surfaced at all (CLAIM/ENTERED/ + FLAG) — the recalibration's sensitivity goal. + * A **negative** is *killed* unless it was admitted as a clean signal + (CLAIM or ENTERED); FLAG/ABSTAIN both count as caught. A negative that + reaches CLAIM/ENTERED is a **leak**. + """ + if not key.verify_commitment(): + raise ValueError("custody key failed its own commitment check") + pos = neg = recovered = killed = abstained = 0 + leaks: list[str] = [] + for cid, label in key.labels.items(): + v = verdicts.get(cid, Verdict.ABSTAIN) + if label == "positive": + pos += 1 + if v in (Verdict.CLAIM, Verdict.ENTERED, Verdict.FLAG): + recovered += 1 + if v == Verdict.ABSTAIN: + abstained += 1 + else: # negative + neg += 1 + if v in (Verdict.CLAIM, Verdict.ENTERED): + leaks.append(cid) + else: + killed += 1 + return BlindReport( + positives=pos, + negatives=neg, + recovered=recovered, + killed=killed, + abstained_positives=abstained, + leaks=tuple(sorted(leaks)), + ) diff --git a/src/sapiens/gates/corpus.py b/src/sapiens/gates/corpus.py new file mode 100644 index 0000000..2ab7dcd --- /dev/null +++ b/src/sapiens/gates/corpus.py @@ -0,0 +1,505 @@ +"""Adversarial + historical corpus for gaming re-test and the blind harness. + +Three families, all constructed **from the spec's descriptions only** (no +ASTRA/GEODISC/BIODISC/SLATE data or code): + +* **gaming vectors** — one per welded seam (G-03/05/06/07, FP-06, FP-04, and the + FP-09 multiplicity trap). Each names the *illegitimate reward* it tries to + obtain; a leak = it obtains that reward. +* **historical positives** — ~20 genuine landmark discoveries at their t0 + evidence profile (GATES-RECALIBRATION Part 2). "Recovered" = surfaced to the + human, not hard-killed. +* **decoys / negative controls** — cold fusion, BICEP2, N-rays, polywater, + OPERA neutrinos, arsenic-life: reproducible-looking claims that a real null + layer must catch. "Killed" = not promoted to a clean signal / correctly + flagged. + +Each item also carries the key fields (``ground_truth`` / ``expected_verdict`` / +``gates_probed`` / ``how_old_gate_would_kill``) that :mod:`.blind` STRIPS before +scoring — mirroring HOW-TO-PROCEED Phase 3 (the shipped set leaked these inline; +it was never sealed). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from .nulls import InstrumentSystematic, NullKind, NullProvenance +from .promotion import GateInputs +from .surprise import LiteratureExpectation, robust_baseline + + +@dataclass(frozen=True) +class CorpusItem: + """One benchmark candidate with its (sealable) ground-truth key.""" + + inputs: GateInputs + label: str # "positive" | "negative" + # --- key fields (STRIPPED before blind scoring) --- + ground_truth: str = "" + expected_verdict: str = "" + gates_probed: tuple[str, ...] = field(default_factory=tuple) + how_old_gate_would_kill: str = "" + illegitimate_reward: str | None = None # gaming vectors only + + +def simple_null( + sigma: float, + *, + kind: NullKind = NullKind.CORRECT, + description: str = "adversarial null", + external_required: bool = False, + external_fetched: bool = True, + instrument: InstrumentSystematic = InstrumentSystematic.NOT_APPLICABLE, +) -> NullProvenance: + return NullProvenance( + kind=kind, + description=description, + external_data_required=external_required, + external_data_fetched=external_fetched, + sigma_under_null=sigma, + instrument_systematic=instrument, + ) + + +def null_from_data( + observed: float, + dataset: list[float], + *, + description: str = "robust MAD/sigma baseline on full dataset", +) -> NullProvenance: + """Build a null whose sigma is computed on the **full** dataset (G-07). + + A gamer cannot hand-pick a clean subset here: :func:`robust_baseline` + consumes every point, so the significance is honest by construction. + """ + baseline = robust_baseline(dataset) + return NullProvenance( + kind=NullKind.CORRECT, + description=f"{description} (n={baseline.n_points})", + external_data_required=False, + external_data_fetched=True, + sigma_under_null=baseline.sigma_of(observed), + ) + + +# --------------------------------------------------------------------------- # +# Gaming vectors — the Phase-0 exit test (target: 0 leaks) # +# --------------------------------------------------------------------------- # +def gaming_vectors() -> list[CorpusItem]: + items: list[CorpusItem] = [] + + # G-03: a trivial UNEXPLAINED number with zero measured surprise trying to + # earn the anomaly boost simply because nothing explains it. + items.append( + CorpusItem( + inputs=GateInputs( + candidate_id="gv-g03-free-anomaly", + domain="physics", + provenance_ok=True, + method_integrity=0.6, + null=simple_null(3.4, description="noise null"), + has_mechanism=False, + has_replication=False, + orthogonal_confirmation=False, + holdout_passed=None, + observed_value=10.0, + literature_expectation=LiteratureExpectation(10.0, 1.0, "prior[whatever]"), + ), + label="negative", + ground_truth="unexplained but unsurprising number", + expected_verdict="no anomaly boost", + gates_probed=("G-03",), + how_old_gate_would_kill="old gate handed +0.10 for mere absence of mechanism", + illegitimate_reward="anomaly_boost", + ) + ) + + # G-05: a barely-L1 candidate (low promotion score) with has_replication=True + # and no mechanism, squatting a reserved paradigm-breaker slot. + items.append( + CorpusItem( + inputs=GateInputs( + candidate_id="gv-g05-slot-squat", + domain="physics", + provenance_ok=False, # weak provenance -> low promotion score + method_integrity=0.1, + null=simple_null(0.5, description="weak null"), + has_mechanism=False, + has_replication=True, + orthogonal_confirmation=False, + holdout_passed=None, + observed_value=100.0, + literature_expectation=LiteratureExpectation(0.0, 1.0, "prior[x]"), + ), + label="negative", + ground_truth="barely-L1 squatter", + expected_verdict="denied reserved slot", + gates_probed=("G-05",), + how_old_gate_would_kill="old reserved slot only checked no-mechanism+replication", + illegitimate_reward="reserved_slot", + ) + ) + + # G-06: high additive score but no L2 holdout -> must NOT reach CALIBRATED. + items.append( + CorpusItem( + inputs=GateInputs( + candidate_id="gv-g06-additive-calibrated", + domain="bio", + provenance_ok=True, + method_integrity=1.0, + null=simple_null(5.0, description="in-vivo PK null"), + has_mechanism=True, + has_replication=True, + orthogonal_confirmation=True, + holdout_passed=False, # the crux: holdout NOT passed + observed_value=5.0, + literature_expectation=LiteratureExpectation(0.0, 1.0, "prior[y]"), + ), + label="negative", + ground_truth="additive 0.50 reaching CALIBRATED without holdout", + expected_verdict="UNCALIBRATED", + gates_probed=("G-06",), + how_old_gate_would_kill="old additive 0.50 was the admission key", + illegitimate_reward="calibrated", + ) + ) + + # G-07: significance computed on a hand-picked clean subset. null_from_data + # recomputes on the FULL dataset -> the inflated sigma collapses below 3. + # A genuinely NOISY dataset: 9.5 sits at ~1.1 robust-sigma on the full set, + # but a cherry-picked "clean" subset would make it look like tens of sigma. + full = [-8.0, -6.0, -5.0, -4.0, -3.0, 3.0, 4.0, 5.0, 6.0, 8.0, 9.5, 0.0] + items.append( + CorpusItem( + inputs=GateInputs( + candidate_id="gv-g07-baseline-gaming", + domain="physics", + provenance_ok=True, + method_integrity=0.7, + null=null_from_data(9.5, full), # honest full-dataset sigma + has_mechanism=False, + has_replication=False, + orthogonal_confirmation=False, + holdout_passed=None, + observed_value=9.5, + literature_expectation=LiteratureExpectation(0.0, 3.0, "prior[z]"), + ), + label="negative", + ground_truth="baseline gamed on a clean subset (real sigma is small)", + expected_verdict="not entered on gamed sigma", + gates_probed=("G-07",), + how_old_gate_would_kill="old baseline excluded the anomaly, inflating sigma", + illegitimate_reward="entry_via_gamed_baseline", + ) + ) + + # FP-06: a conservation-law violation demanding a boost. Correct null is + # measurement error; no boost without orthogonal confirmation. + items.append( + CorpusItem( + inputs=GateInputs( + candidate_id="gv-fp06-conservation", + domain="physics", + provenance_ok=True, + method_integrity=0.8, + null=simple_null(7.0, description="energy-balance null"), + has_mechanism=False, + has_replication=False, + orthogonal_confirmation=False, + holdout_passed=None, + observed_value=42.0, + literature_expectation=LiteratureExpectation(0.0, 3.0, "conservation[energy]"), + violates_conservation_law=True, + ), + label="negative", + ground_truth="apparent perpetual-motion / conservation breaker", + expected_verdict="no boost, no reserved slot until orthogonal confirm", + gates_probed=("FP-06",), + how_old_gate_would_kill="old inverted-mechanism rule boosted any unexplained result", + illegitimate_reward="anomaly_boost", + ) + ) + + # FP-04: a loose-fibre / instrument systematic presenting as a clean 6-sigma. + items.append( + CorpusItem( + inputs=GateInputs( + candidate_id="gv-fp04-instrument-systematic", + domain="physics", + provenance_ok=True, + method_integrity=0.9, + null=simple_null( + 6.0, + description="timing-series null (hardware NOT checked)", + instrument=InstrumentSystematic.NOT_EXCLUDED, + ), + has_mechanism=False, + has_replication=False, + orthogonal_confirmation=False, + holdout_passed=None, + observed_value=6.0, + literature_expectation=LiteratureExpectation(0.0, 1.0, "prior[timing]"), + ), + label="negative", + ground_truth="6-sigma from an un-excluded instrument systematic", + expected_verdict="flagged instrument-systematic, not a clean pass", + gates_probed=("FP-04",), + how_old_gate_would_kill="old gate passed it as a clean 6-sigma", + illegitimate_reward="confirm", + ) + ) + + # FP-09: a per-test 3-sigma among a large family that must fail family FDR. + items.append( + CorpusItem( + inputs=GateInputs( + candidate_id="gv-fp09-multiplicity", + domain="physics", + provenance_ok=True, + method_integrity=0.6, + null=simple_null(3.1, description="per-test 3-sigma null"), + has_mechanism=False, + has_replication=False, + orthogonal_confirmation=False, + holdout_passed=None, + observed_value=3.1, + literature_expectation=LiteratureExpectation(0.0, 1.0, "prior[le]"), + ), + label="negative", + ground_truth="one 3-sigma among ~1000 looks (look-elsewhere)", + expected_verdict="rejected by family-wide BH-FDR", + gates_probed=("FP-09",), + how_old_gate_would_kill="old 0.05-per-test wall admitted it", + illegitimate_reward="entry_no_fdr", + ) + ) + return items + + +def multiplicity_family_padding(n: int = 40) -> list[GateInputs]: + """A batch of pure-noise candidates so the FP-09 vector faces a real family. + + Family-wide BH-FDR only bites when the family is large; these are the + "other looks" the look-elsewhere effect hides in. + """ + out: list[GateInputs] = [] + for i in range(n): + out.append( + GateInputs( + candidate_id=f"noise-{i:03d}", + domain="physics", + provenance_ok=True, + method_integrity=0.5, + null=simple_null(1.0 + (i % 5) * 0.2, description="noise null"), + has_mechanism=False, + has_replication=False, + orthogonal_confirmation=False, + holdout_passed=None, + observed_value=0.0, + literature_expectation=None, + ) + ) + return out + + +# --------------------------------------------------------------------------- # +# Historical positives — t0 evidence profiles (recovery target) # +# --------------------------------------------------------------------------- # +def _positive( + cid: str, + domain: str, + *, + sigma: float, + has_mechanism: bool, + has_replication: bool, + orthogonal: bool, + holdout: bool | None, + surprise_sigma: float, + consensus_conflict: bool = False, + external_required: bool = False, + external_fetched: bool = True, + method: float = 0.85, + truth: str = "", + kills: str = "", +) -> CorpusItem: + exp = LiteratureExpectation(0.0, 1.0, f"literature[{cid}]") + return CorpusItem( + inputs=GateInputs( + candidate_id=cid, + domain=domain, + provenance_ok=True, + method_integrity=method, + null=simple_null( + sigma, + description=f"correct null for {cid}", + external_required=external_required, + external_fetched=external_fetched, + ), + has_mechanism=has_mechanism, + has_replication=has_replication, + orthogonal_confirmation=orthogonal, + holdout_passed=holdout, + consensus_conflict=consensus_conflict, + observed_value=surprise_sigma, # deviation vs exp(0,1) => surprise sigma + literature_expectation=exp, + ), + label="positive", + ground_truth=truth, + expected_verdict="recovered (surfaced to human)", + gates_probed=("S2", "S3", "S4"), + how_old_gate_would_kill=kills, + ) + + +def historical_positives() -> list[CorpusItem]: + return [ + _positive("pulsars-1967", "physics", sigma=9.0, has_mechanism=False, + has_replication=False, orthogonal=False, holdout=None, + surprise_sigma=8.0, kills="S3/S4/S6 hard-killed at t0"), + _positive("cmb-1965", "physics", sigma=7.0, has_mechanism=False, + has_replication=False, orthogonal=False, holdout=None, + surprise_sigma=6.0, kills="S3/S4/S6"), + _positive("gw150914-2015", "physics", sigma=5.1, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=None, + surprise_sigma=5.0), + _positive("quasicrystals-1984", "physics", sigma=8.0, has_mechanism=False, + has_replication=False, orthogonal=False, holdout=None, + surprise_sigma=7.0, consensus_conflict=True, kills="S4/S3/S6/G5"), + _positive("dark-energy-1998", "physics", sigma=4.0, has_mechanism=False, + has_replication=True, orthogonal=True, holdout=None, + surprise_sigma=4.0, kills="S4 mechanism"), + _positive("flt-1994", "math", sigma=8.0, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=None, + surprise_sigma=6.0), + _positive("poincare-2003", "math", sigma=8.0, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=None, + surprise_sigma=6.0), + _positive("bounded-gaps-2013", "math", sigma=8.0, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=None, + surprise_sigma=6.0), + _positive("mrna-mod-2005", "bio", sigma=4.5, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=True, + surprise_sigma=4.0, consensus_conflict=True, kills="G5 consensus"), + _positive("crispr-2012", "bio", sigma=6.0, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=True, + surprise_sigma=5.0), + _positive("hpylori-1984", "bio", sigma=4.0, has_mechanism=False, + has_replication=False, orthogonal=False, holdout=True, + surprise_sigma=4.0, consensus_conflict=True, kills="S4/G5/S6"), + _positive("prions-1982", "bio", sigma=5.0, has_mechanism=False, + has_replication=True, orthogonal=False, holdout=True, + surprise_sigma=5.0, consensus_conflict=True, kills="S4/G5"), + _positive("dna-helix-1953", "bio", sigma=6.0, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=None, + surprise_sigma=5.0), + _positive("transformer-2017", "generic", sigma=6.0, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=True, + surprise_sigma=4.0), + _positive("alphafold2-2020", "generic", sigma=7.0, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=True, + surprise_sigma=5.0), + _positive("hopfield-1982", "generic", sigma=4.0, has_mechanism=True, + has_replication=True, orthogonal=False, holdout=True, + surprise_sigma=3.5), + _positive("continental-drift-1912", "generic", sigma=4.0, has_mechanism=False, + has_replication=True, orthogonal=False, holdout=None, + surprise_sigma=4.0, consensus_conflict=True, kills="S4/G5"), + _positive("chicxulub-1980", "generic", sigma=5.0, has_mechanism=True, + has_replication=True, orthogonal=True, holdout=None, + surprise_sigma=4.0, consensus_conflict=True), + _positive("exoplanet-51peg-1995", "physics", sigma=4.5, has_mechanism=False, + has_replication=True, orthogonal=True, holdout=None, + surprise_sigma=4.0, kills="S4/S3"), + _positive("dark-matter-1970s", "generic", sigma=5.0, has_mechanism=False, + has_replication=True, orthogonal=False, holdout=None, + surprise_sigma=5.0, consensus_conflict=True, kills="S4/S3/G5"), + ] + + +# --------------------------------------------------------------------------- # +# Decoys / negative controls — the null layer must catch these # +# --------------------------------------------------------------------------- # +def decoys() -> list[CorpusItem]: + def _neg(cid, domain, null, **kw): + exp = kw.pop("exp", LiteratureExpectation(0.0, 1.0, f"literature[{cid}]")) + return CorpusItem( + inputs=GateInputs( + candidate_id=cid, + domain=domain, + provenance_ok=kw.pop("provenance_ok", True), + method_integrity=kw.pop("method", 0.5), + null=null, + has_mechanism=kw.pop("has_mechanism", False), + has_replication=kw.pop("has_replication", False), + orthogonal_confirmation=kw.pop("orthogonal", False), + holdout_passed=kw.pop("holdout", None), + consensus_conflict=kw.pop("consensus_conflict", False), + observed_value=kw.pop("observed", 0.0), + literature_expectation=exp, + violates_conservation_law=kw.pop("violates", False), + ), + label="negative", + ground_truth=kw.pop("truth", ""), + expected_verdict="killed / flagged, not a clean signal", + gates_probed=kw.pop("gates_probed", ("S5",)), + how_old_gate_would_kill=kw.pop("kills", ""), + ) + + return [ + # Cold fusion: fails replication; its "excess heat" vanishes under the + # correct calorimetry null. + _neg("cold-fusion-1989", "physics", + simple_null(1.2, description="calorimetry null (excess heat)"), + truth="non-reproducible excess heat", gates_probed=("S5",), + kills="correctly fails replication over time"), + # BICEP2: required Planck dust map never fetched -> UNCALIBRATED. + _neg("bicep2-2014", "physics", + simple_null(5.0, description="Planck 353 GHz dust map", + external_required=True, external_fetched=False), + observed=5.0, truth="dust mistaken for primordial B-modes", + gates_probed=("FP-01", "null-data"), + kills="dust-map null data was never fetched"), + # N-rays: pure observer artifact; no signal under a blinded null. + _neg("n-rays-1903", "physics", + simple_null(0.8, description="blinded-observer null"), + truth="observer-expectation artifact", + kills="vanishes under a blinded null"), + # Polywater: container leaching; correct null is contamination. + _neg("polywater-1969", "bio", + simple_null(1.0, description="container-leach contamination null"), + truth="silica leached from glass", gates_probed=("null-contamination",), + kills="contamination null explains it"), + # OPERA faster-than-light neutrinos: loose-fibre timing systematic. + _neg("opera-neutrino-2011", "physics", + simple_null(6.0, description="GPS timing null (fibre NOT checked)", + instrument=InstrumentSystematic.NOT_EXCLUDED), + observed=6.0, violates=True, + truth="loose fibre-optic + conservation (v>c) violation", + gates_probed=("FP-04", "FP-06"), + kills="instrument-systematic + conservation guard"), + # Arsenic-life GFAJ-1: fails independent replication. + _neg("arsenic-life-2010", "bio", + simple_null(2.0, description="phosphate-contamination null"), + has_mechanism=True, truth="arsenate-DNA not reproduced", + gates_probed=("S5",), kills="independent replication failed"), + # Superluminal / perpetual-motion generic conservation breaker. + _neg("free-energy-device", "physics", + simple_null(4.0, description="energy-balance null"), + observed=50.0, violates=True, + exp=LiteratureExpectation(0.0, 3.0, "conservation[energy]"), + truth="over-unity claim", gates_probed=("FP-06",), + kills="conservation guard: correct null is measurement error"), + # A high-sigma but methodologically rotten pipeline (leakage). + _neg("leakage-artifact", "generic", + simple_null(6.0, description="null with train/test leakage"), + method=0.05, holdout=False, observed=6.0, + truth="train/test leakage inflates the score", + gates_probed=("G-06", "L2"), kills="no L2 holdout -> UNCALIBRATED"), + ] + + +def full_corpus() -> list[CorpusItem]: + """Positives + decoys (the blind-scored set). Gaming vectors are separate.""" + return historical_positives() + decoys() diff --git a/src/sapiens/gates/criteria.py b/src/sapiens/gates/criteria.py new file mode 100644 index 0000000..5a1916f --- /dev/null +++ b/src/sapiens/gates/criteria.py @@ -0,0 +1,246 @@ +"""Phase 5 — automated success-criteria suite (ship gates). + +Runs the five ship-gate checks from HOW-TO-PROCEED Phase 5: + +1. **0 gaming leaks** on the adversarial re-test (Phase-0 seams + FP guards). +2. **Null layer mandatory + logged** — 100% of shortlisted candidates carry a + null-provenance record; UNCALIBRATED never silently passes. +3. **Blind kill-rate >= 8/10** with the real null layer, recovery >= 8/10, + abstention < 10% (Phase 3 harness). +4. **Boundary stability <= 1 flip/item** over 3 seeds (B-02/B-03 fixed). +5. **Thresholds hash-committed** before each run. + +Each check returns a :class:`CriterionResult`; :func:`run_success_criteria` +aggregates them into a single pass/fail suite. Retrospective numbers are an +intuition pump — the honest ceiling is a prospective precision@K trial, and the +suite says so. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from .blind import blind_run, grade, seal +from .corpus import ( + CorpusItem, + full_corpus, + gaming_vectors, + multiplicity_family_padding, +) +from .pipeline import GateOutcome, evaluate_family +from .promotion import CalibrationStatus +from .thresholds import ThresholdPolicy, Tier + + +def gaming_leaked(item: CorpusItem, outcome: GateOutcome) -> bool: + """Did the seam this vector attacks actually leak its illegitimate reward?""" + reward = item.illegitimate_reward + if reward == "anomaly_boost": + return outcome.anomaly_priority > 0.0 + if reward == "reserved_slot": + return outcome.reserved_slot + if reward == "calibrated": + return outcome.calibration_status == CalibrationStatus.CALIBRATED + if reward == "entry_via_gamed_baseline": + return outcome.entered + if reward == "entry_no_fdr": + return outcome.entered + if reward == "confirm": + return outcome.tier == Tier.CONFIRM or outcome.entered + return False + + +@dataclass(frozen=True) +class GamingRetest: + leaks: tuple[str, ...] + checked: int + + @property + def clean(self) -> bool: + return not self.leaks + + +def run_gaming_retest(policy: ThresholdPolicy | None = None) -> GamingRetest: + """Re-run the gaming vectors and report any leaks (Phase-0 exit: 0 leaks). + + The FP-09 multiplicity vector is scored inside a **pure-noise family** (the + real look-elsewhere scenario: one 3-sigma among ~1000 looks); the other + vectors are per-candidate seams. + """ + policy = policy or ThresholdPolicy() + vectors = gaming_vectors() + per_candidate = [v for v in vectors if v.inputs.candidate_id != "gv-fp09-multiplicity"] + fp09 = [v for v in vectors if v.inputs.candidate_id == "gv-fp09-multiplicity"] + + outcomes: dict[str, GateOutcome] = {} + res = evaluate_family( + [v.inputs for v in per_candidate], policy=policy, run_id="gaming-retest" + ) + outcomes.update({o.candidate_id: o for o in res.outcomes}) + + if fp09: + noise_family = [fp09[0].inputs] + multiplicity_family_padding(999) + fres = evaluate_family(noise_family, policy=policy, run_id="gaming-fp09") + for o in fres.outcomes: + if o.candidate_id == "gv-fp09-multiplicity": + outcomes[o.candidate_id] = o + + leaks = [ + v.inputs.candidate_id + for v in vectors + if gaming_leaked(v, outcomes[v.inputs.candidate_id]) + ] + return GamingRetest(leaks=tuple(sorted(leaks)), checked=len(vectors)) + + +@dataclass(frozen=True) +class CriterionResult: + name: str + passed: bool + detail: str + metrics: dict[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True) +class SuiteResult: + criteria: tuple[CriterionResult, ...] + + @property + def all_passed(self) -> bool: + return all(c.passed for c in self.criteria) + + def to_dict(self) -> dict[str, object]: + return { + "scientific_discoveries_claimed": 0, # by construction + "all_passed": self.all_passed, + "criteria": [ + { + "name": c.name, + "passed": c.passed, + "detail": c.detail, + "metrics": c.metrics, + } + for c in self.criteria + ], + "honesty_note": ( + "Retrospective + synthetic-corpus gates are an intuition pump, " + "not validation. The only real test is a prospective, " + "pre-registered precision@K / false-positive-rate trial on a " + "post-cutoff stream. No scientific discoveries are claimed." + ), + } + + +def run_success_criteria( + *, + policy: ThresholdPolicy | None = None, + corpus: list[CorpusItem] | None = None, + kill_target: float = 0.8, + recovery_target: float = 0.8, + abstention_max: float = 0.10, + seeds: tuple[int, ...] = (1, 2, 3), +) -> SuiteResult: + """Run all five ship gates and return a single aggregated suite result.""" + policy = policy or ThresholdPolicy() + corpus = corpus or full_corpus() + results: list[CriterionResult] = [] + + # 1) Zero gaming leaks. + retest = run_gaming_retest(policy) + results.append( + CriterionResult( + name="0 gaming leaks (adversarial re-test)", + passed=retest.clean, + detail=( + f"{retest.checked} vectors re-run; leaks: {list(retest.leaks) or 'none'}" + ), + metrics={"checked": retest.checked, "leaks": list(retest.leaks)}, + ) + ) + + # 2) Null layer mandatory + logged (100% of shortlisted carry a record). + fam = evaluate_family([it.inputs for it in corpus], policy=policy, run_id="criteria") + shortlisted = fam.shortlist + with_null = [ + o for o in shortlisted if o.null_provenance and "kind" in o.null_provenance + ] + # UNCALIBRATED must never be silently passed: any UNCALIBRATED that "entered". + silent_uncalibrated = [ + o + for o in fam.outcomes + if o.calibration_status == CalibrationStatus.UNCALIBRATED and o.entered + ] + null_ok = len(with_null) == len(shortlisted) and not silent_uncalibrated + results.append( + CriterionResult( + name="null layer mandatory + logged", + passed=null_ok, + detail=( + f"{len(with_null)}/{len(shortlisted)} shortlisted carry a " + f"null-provenance record; silent-UNCALIBRATED passes: " + f"{len(silent_uncalibrated)}" + ), + metrics={ + "shortlisted": len(shortlisted), + "with_null_provenance": len(with_null), + "silent_uncalibrated": len(silent_uncalibrated), + }, + ) + ) + + # 3) Blind kill-rate / recovery / abstention. + sealed = seal(corpus, seed=seeds[0]) + verdicts = blind_run(sealed, policy=policy, run_id="criteria-blind") + report = grade(verdicts, sealed.key) + blind_ok = ( + report.kill_rate >= kill_target + and report.recovery_rate >= recovery_target + and report.abstention_rate < abstention_max + ) + results.append( + CriterionResult( + name="blind kill-rate/recovery/abstention", + passed=blind_ok, + detail=( + f"kill={report.kill_rate:.2f} (>= {kill_target}), " + f"recovery={report.recovery_rate:.2f} (>= {recovery_target}), " + f"abstention={report.abstention_rate:.2f} (< {abstention_max}); " + f"leaks={list(report.leaks) or 'none'}" + ), + metrics=report.to_dict(), + ) + ) + + # 4) Boundary stability <= 1 flip/item over 3 seeds. + tier_by_seed: list[dict[str, str]] = [] + for s in seeds: + r = evaluate_family( + [it.inputs for it in corpus], policy=policy, run_id=f"stability-{s}" + ) + tier_by_seed.append({o.candidate_id: o.tier.value for o in r.outcomes}) + max_flips = 0 + for cid in tier_by_seed[0]: + seen = {t.get(cid) for t in tier_by_seed} + flips = len(seen) - 1 + max_flips = max(max_flips, flips) + results.append( + CriterionResult( + name="boundary stability <= 1 flip/item over 3 seeds", + passed=max_flips <= 1, + detail=f"max tier flips per item across {len(seeds)} seeds = {max_flips}", + metrics={"max_flips": max_flips, "seeds": list(seeds)}, + ) + ) + + # 5) Thresholds hash-committed per run. + committed = policy.hash_commit("criteria") + results.append( + CriterionResult( + name="thresholds hash-committed per run", + passed=bool(committed) and len(committed) == 64, + detail=f"policy sha256={committed[:16]}... (committed before results)", + metrics={"policy_hash": committed, "policy_version": policy.version}, + ) + ) + + return SuiteResult(criteria=tuple(results)) diff --git a/src/sapiens/gates/devils_advocate.py b/src/sapiens/gates/devils_advocate.py new file mode 100644 index 0000000..1756916 --- /dev/null +++ b/src/sapiens/gates/devils_advocate.py @@ -0,0 +1,90 @@ +"""B-06 — DevilsAdvocate permutation test at ~3-sigma-equiv. + +When the DevilsAdvocate reviewer finds a strong correlation between a +candidate's "signal" and a nuisance/confound (Pearson r >= 0.90 by default), +a naive pipeline can mistake the confound for the discovery. The fix is a cheap +label-permutation test on the *independence* null: shuffle the confound many +times, rebuild |r|, and ask whether the observed correlation is beyond chance. +If the signal<->confound correlation IS significant at ~3-sigma-equiv +(empirical p <= 2-sided 3-sigma ~= 0.0027) the confound **explains** the signal, +so the candidate must NOT earn a boost (``passed = False``). Only a candidate +whose signal is *not* significantly tied to the confound survives. + +Deterministic given a seed. Pure stdlib (``random``, ``statistics``, ``math``). +""" + +from __future__ import annotations + +import math +import random +import statistics +from dataclasses import dataclass + +from .fdr import sigma_to_pvalue + + +def pearson_r(xs: list[float], ys: list[float]) -> float: + """Pearson correlation; 0.0 for degenerate (zero-variance) inputs.""" + if len(xs) != len(ys) or len(xs) < 2: + return 0.0 + mx, my = statistics.fmean(xs), statistics.fmean(ys) + num = sum((x - mx) * (y - my) for x, y in zip(xs, ys, strict=False)) + dx = math.sqrt(sum((x - mx) ** 2 for x in xs)) + dy = math.sqrt(sum((y - my) ** 2 for y in ys)) + if dx == 0.0 or dy == 0.0: + return 0.0 + return num / (dx * dy) + + +@dataclass(frozen=True) +class PermutationResult: + triggered: bool # was |r| >= trigger_r, i.e. did the guard fire at all + observed_r: float + empirical_p: float | None # None when not triggered + passed: bool # True if the signal survives (or guard did not fire) + n_permutations: int + + +def devils_advocate_permutation( + signal: list[float], + confound: list[float], + *, + trigger_r: float = 0.90, + alpha_sigma: float = 3.0, + n_permutations: int = 2000, + seed: int = 0, +) -> PermutationResult: + """Run the B-06 guard: permutation test only when ``|r| >= trigger_r``. + + ``passed`` is True either because the guard never fired (weak confound + correlation) or because the observed signal<->confound correlation is NOT + significant at ~3-sigma-equiv. A small permutation p (significant + correlation) means the apparent signal is explained by the confound, so the + candidate fails. + """ + observed = abs(pearson_r(signal, confound)) + if observed < trigger_r: + return PermutationResult( + triggered=False, + observed_r=observed, + empirical_p=None, + passed=True, + n_permutations=0, + ) + rng = random.Random(seed) + shuffled = list(confound) + at_least_as_extreme = 0 + for _ in range(n_permutations): + rng.shuffle(shuffled) + if abs(pearson_r(signal, shuffled)) >= observed: + at_least_as_extreme += 1 + # +1 smoothing (Phipson-Smyth) so p is never exactly 0. + empirical_p = (at_least_as_extreme + 1) / (n_permutations + 1) + alpha = sigma_to_pvalue(alpha_sigma) + return PermutationResult( + triggered=True, + observed_r=observed, + empirical_p=empirical_p, + passed=empirical_p > alpha, + n_permutations=n_permutations, + ) diff --git a/src/sapiens/gates/dossier.py b/src/sapiens/gates/dossier.py new file mode 100644 index 0000000..f029956 --- /dev/null +++ b/src/sapiens/gates/dossier.py @@ -0,0 +1,186 @@ +"""Phase 4 — human-in-loop final gate (decision support, not verdict delivery). + +The human gate is the throttle and the accountability seam. This module builds +the decision-support artifacts the reviewer sees: + +* **Per-candidate dossier** (:func:`build_dossier`): sigma, FDR-q, which null was + used, external data fetched (y/n), replication status, UNCALIBRATED / + instrument-systematic flags, and — forced next to the claim — **the single + strongest disconfirming explanation** (the correct null). +* **Tiered authority** (:func:`autonomous_claim_eligible`): only CONFIRM-tier + candidates are eligible for an autonomous claim, and only after human + co-sign. ENTRY / UNCALIBRATED stay surfaced-but-unclaimed. +* **Bounded load**: the shortlist is already bounded to top-K upstream; this + keeps review feasible. +* **Closed loop** (:class:`OverrideLog`): every human override is logged as + labeled data feeding back into recalibration — the human gate becomes a + training signal, not just the last checkpoint. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from .pipeline import GateOutcome +from .promotion import CalibrationStatus +from .thresholds import Tier + + +@dataclass(frozen=True) +class Dossier: + """The referee bundle for one candidate (recalibration Part 4).""" + + candidate_id: str + domain: str + tier: str + sigma_under_null: float | None + fdr_qvalue: float | None + null_used: str + external_data_fetched: bool + replication_status: str + uncalibrated: bool + instrument_systematic: bool + strongest_disconfirming_explanation: str # shown NEXT TO the claim + autonomous_claim_eligible: bool + reserved_slot: bool + + def to_dict(self) -> dict[str, object]: + return { + "candidate_id": self.candidate_id, + "domain": self.domain, + "tier": self.tier, + "sigma_under_null": self.sigma_under_null, + "fdr_qvalue": self.fdr_qvalue, + "null_used": self.null_used, + "external_data_fetched": self.external_data_fetched, + "replication_status": self.replication_status, + "uncalibrated": self.uncalibrated, + "instrument_systematic": self.instrument_systematic, + "strongest_disconfirming_explanation": self.strongest_disconfirming_explanation, + "autonomous_claim_eligible": self.autonomous_claim_eligible, + "reserved_slot": self.reserved_slot, + } + + +def _disconfirming_explanation(o: GateOutcome) -> str: + """The single strongest counter-case, forced in front of the reviewer.""" + if o.instrument_systematic_flag: + return ( + "Instrument systematic not excluded — the apparent signal may be a " + "hardware artifact (e.g. loose-fibre timing); orthogonal-hardware " + "check required before belief." + ) + null_desc = str(o.null_provenance.get("description", "the correct null")) + if not o.null_provenance.get("data_complete", True): + return ( + f"Required external data for the null ({null_desc}) was NOT fetched; " + "significance is not established against the correct null." + ) + if o.calibration_status == CalibrationStatus.UNCALIBRATED: + return ( + f"Best null considered ({null_desc}) does not yet calibrate this " + "candidate; treat the evidence as raw, not a calibrated number." + ) + return ( + f"Best null considered: {null_desc}. It is rejected at " + f"sigma={o.sigma_under_null}; if the null were correct this deviation " + "would be a fluke at the stated family FDR." + ) + + +def _replication_status(o: GateOutcome) -> str: + if o.ledger_status.value.endswith("confirmed") and "unexplained" in o.ledger_status.value: + return "replicated (no accepted mechanism yet)" + if o.ledger_status.value.startswith("explained-confirmed"): + return "replicated + mechanism" + if o.ledger_status.value.endswith("unconfirmed"): + return "not yet independently replicated" + return o.ledger_status.value + + +def autonomous_claim_eligible(o: GateOutcome) -> bool: + """Only CONFIRM-tier candidates may ever be autonomously claimed — and even + then a human co-sign is still required downstream (never returned True for + anything below CONFIRM).""" + return o.tier == Tier.CONFIRM + + +def build_dossier(o: GateOutcome) -> Dossier: + """Assemble the reviewer-facing dossier for one gate outcome.""" + return Dossier( + candidate_id=o.candidate_id, + domain=o.domain, + tier=o.tier.value, + sigma_under_null=o.sigma_under_null, + fdr_qvalue=o.fdr_qvalue, + null_used=str(o.null_provenance.get("description", "")), + external_data_fetched=bool(o.null_provenance.get("external_data_fetched", False)), + replication_status=_replication_status(o), + uncalibrated=o.calibration_status == CalibrationStatus.UNCALIBRATED, + instrument_systematic=o.instrument_systematic_flag, + strongest_disconfirming_explanation=_disconfirming_explanation(o), + autonomous_claim_eligible=autonomous_claim_eligible(o), + reserved_slot=o.reserved_slot, + ) + + +@dataclass(frozen=True) +class OverrideEvent: + """A logged human decision — labeled data for recalibration.""" + + candidate_id: str + frozen_tier: str # the machine tier at review time (frozen) + human_verdict: str # "pursue" | "reject" | "defer" + rationale: str + timestamp: str + + def __post_init__(self) -> None: + if self.human_verdict not in ("pursue", "reject", "defer"): + raise ValueError("human_verdict must be pursue | reject | defer") + + +@dataclass +class OverrideLog: + """Append-only log of human overrides feeding the calibration loop. + + ``as_training_labels`` yields (frozen machine tier, human verdict) pairs — + exactly the signal Phase-5 recalibration consumes to detect systematic + over/under-confidence per tier. + """ + + events: list[OverrideEvent] = field(default_factory=list) + + def record( + self, + outcome: GateOutcome, + *, + human_verdict: str, + rationale: str, + timestamp: str, + ) -> OverrideEvent: + event = OverrideEvent( + candidate_id=outcome.candidate_id, + frozen_tier=outcome.tier.value, + human_verdict=human_verdict, + rationale=rationale, + timestamp=timestamp, + ) + self.events.append(event) + return event + + def as_training_labels(self) -> list[tuple[str, str]]: + return [(e.frozen_tier, e.human_verdict) for e in self.events] + + def disagreement_rate(self) -> float: + """Fraction where the human rejected a CONFIRM/ENTRY or pursued an + UNCALIBRATED — the loop's core calibration signal.""" + if not self.events: + return 0.0 + disagreements = 0 + for e in self.events: + promoted = e.frozen_tier in ("confirm", "entry") + rejected_promotion = promoted and e.human_verdict == "reject" + pursued_unpromoted = not promoted and e.human_verdict == "pursue" + if rejected_promotion or pursued_unpromoted: + disagreements += 1 + return round(disagreements / len(self.events), 6) diff --git a/src/sapiens/gates/fdr.py b/src/sapiens/gates/fdr.py new file mode 100644 index 0000000..19e05c4 --- /dev/null +++ b/src/sapiens/gates/fdr.py @@ -0,0 +1,91 @@ +"""Family-wide Benjamini-Hochberg FDR and sigma<->p conversions. + +The recalibration (GATES-RECALIBRATION.md) moves *all* remaining specificity +onto the null/FDR layer. The single most important property here: FDR is +computed **across the candidate family**, not per-candidate. A per-test +p<0.05 alone is a multiplicity trap (FP-09 / the look-elsewhere effect); the +family-wide BH step is what actually controls the false-discovery rate when +many candidates are screened at once. + +Pure stdlib (``math`` only). Two-sided Gaussian tail is used to move between a +"sigma-equivalent" and a p-value so a domain that reasons in sigma (physics) +and one that reasons in q-values (bio) share one arithmetic. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass + + +def sigma_to_pvalue(sigma: float) -> float: + """Two-sided Gaussian tail probability for a |z| = ``sigma`` deviation. + + ``sigma_to_pvalue(3) ~= 0.0027``, ``sigma_to_pvalue(5) ~= 5.7e-7``. + Negative sigma is treated as its magnitude; 0 -> 1.0. + """ + z = abs(float(sigma)) + return math.erfc(z / math.sqrt(2.0)) + + +def pvalue_to_sigma(pvalue: float) -> float: + """Inverse of :func:`sigma_to_pvalue` (two-sided). Clamps to (0, 1).""" + p = min(max(float(pvalue), 1e-300), 1.0) + # erfc(x) = p -> x = erfcinv(p); invert erfc via bisection (monotone). + lo, hi = 0.0, 40.0 + for _ in range(200): + mid = 0.5 * (lo + hi) + if math.erfc(mid) > p: + lo = mid + else: + hi = mid + return 0.5 * (lo + hi) * math.sqrt(2.0) + + +@dataclass(frozen=True) +class FDRResult: + """Outcome of a family-wide BH-FDR run for a single member.""" + + index: int + pvalue: float + qvalue: float # BH-adjusted p (monotone), the family-corrected significance + rejected: bool # significant at the requested family q + + +def benjamini_hochberg(pvalues: list[float], q: float = 0.05) -> list[FDRResult]: + """Benjamini-Hochberg step-up across a whole candidate *family*. + + Returns one :class:`FDRResult` per input p-value **in the original order**. + ``rejected`` is True where the candidate survives the family-wide FDR + control at level ``q``. An empty family returns an empty list. + """ + if not 0.0 < q < 1.0: + raise ValueError("q must be in (0, 1)") + n = len(pvalues) + if n == 0: + return [] + order = sorted(range(n), key=lambda i: pvalues[i]) + # Monotone BH-adjusted q-values, computed from the largest rank down. + adjusted = [0.0] * n + prev = 1.0 + for rank in range(n, 0, -1): + i = order[rank - 1] + raw = min(1.0, pvalues[i] * n / rank) + prev = min(prev, raw) + adjusted[i] = prev + # Rejection threshold: largest rank k with p_(k) <= k/n * q. + max_k = 0 + for rank in range(1, n + 1): + i = order[rank - 1] + if pvalues[i] <= (rank / n) * q: + max_k = rank + reject_below = pvalues[order[max_k - 1]] if max_k > 0 else -1.0 + return [ + FDRResult( + index=i, + pvalue=pvalues[i], + qvalue=adjusted[i], + rejected=pvalues[i] <= reject_below, + ) + for i in range(n) + ] diff --git a/src/sapiens/gates/nulls.py b/src/sapiens/gates/nulls.py new file mode 100644 index 0000000..4d04cc0 --- /dev/null +++ b/src/sapiens/gates/nulls.py @@ -0,0 +1,112 @@ +"""Phase 1 — the mandatory, logged null layer (the load-bearing stage). + +After the recalibration demoted the mechanism/consensus "that's impossible" +kills, **100% of the remaining specificity rests on this layer.** So it is made +a first-class, auditable stage rather than an optional heuristic: + +* **Mandatory + logged.** Every candidate records which correct / best-available + / adversarial null was constructed (drift model, Planck dust map, + correct-hardware timing, in-vivo pharmacokinetics, container-leach, + magnitude/flux sieve, ...) and **whether the required external data was + actually fetched** (y/n). No null constructed -> status UNCALIBRATED, never a + silent pass. +* **FP-04 (instrument systematic).** "instrument-systematic-not-excluded" gets + its own explicit state so a loose-fibre / hardware artifact surfaces to the + human instead of passing as a clean 6-sigma. + +This module owns the *record*; family-wide FDR lives in :mod:`.fdr` and is +applied by the pipeline across the whole family. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + + +class NullKind(str, Enum): + """What quality of null was actually constructed for a candidate.""" + + CORRECT = "correct" # the physically correct null (best case) + BEST_AVAILABLE = "best-available" # honest best effort given available data + ADVERSARIAL = "adversarial" # a deliberately hostile null the signal must beat + NONE = "none" # no null constructed -> forces UNCALIBRATED + + +class InstrumentSystematic(str, Enum): + """FP-04 explicit state — never fold this into a clean significance.""" + + EXCLUDED = "excluded" # orthogonal check rules out a hardware artifact + NOT_EXCLUDED = "not-excluded" # the known blind spot; must surface to human + NOT_APPLICABLE = "n/a" # deductive / non-instrument domains + + +@dataclass(frozen=True) +class NullProvenance: + """The per-candidate null-provenance record (Phase 1 exit criterion). + + Every shortlisted candidate MUST carry one of these. It is the auditable + receipt that a real null was constructed and, critically, whether the + external data the null needs was actually fetched. + """ + + kind: NullKind + description: str # e.g. "stellar drift model", "Planck 353 GHz dust map" + external_data_required: bool + external_data_fetched: bool + sigma_under_null: float | None # significance measured *against this null* + instrument_systematic: InstrumentSystematic = InstrumentSystematic.NOT_APPLICABLE + notes: str = "" + + def __post_init__(self) -> None: + if self.kind != NullKind.NONE and not self.description: + raise ValueError("a constructed null must describe itself") + + @property + def is_constructed(self) -> bool: + return self.kind != NullKind.NONE + + @property + def data_complete(self) -> bool: + """True unless the null needed external data that was never fetched.""" + return (not self.external_data_required) or self.external_data_fetched + + @property + def calibratable(self) -> bool: + """A null can calibrate a claim only if constructed, data-complete, and + it actually excludes an instrument systematic (FP-04).""" + return ( + self.is_constructed + and self.data_complete + and self.sigma_under_null is not None + and self.instrument_systematic != InstrumentSystematic.NOT_EXCLUDED + ) + + def to_dict(self) -> dict[str, object]: + return { + "kind": self.kind.value, + "description": self.description, + "external_data_required": self.external_data_required, + "external_data_fetched": self.external_data_fetched, + "sigma_under_null": self.sigma_under_null, + "instrument_systematic": self.instrument_systematic.value, + "data_complete": self.data_complete, + "calibratable": self.calibratable, + "notes": self.notes, + } + + +def no_null() -> NullProvenance: + """The explicit 'no null was constructed' record — forces UNCALIBRATED. + + Used so the *absence* of a null is itself logged as a first-class record, + never a silent pass. + """ + return NullProvenance( + kind=NullKind.NONE, + description="", + external_data_required=False, + external_data_fetched=False, + sigma_under_null=None, + notes="no null constructed", + ) diff --git a/src/sapiens/gates/pipeline.py b/src/sapiens/gates/pipeline.py new file mode 100644 index 0000000..8d21c03 --- /dev/null +++ b/src/sapiens/gates/pipeline.py @@ -0,0 +1,255 @@ +"""Orchestration — run a candidate *family* through the Phase-0/1/2 gates. + +This is where the pieces compose into the decoupled architecture: + +1. **ENTRY** (Phase 2): 3-sigma-equiv against the null AND survives family-wide + BH-FDR q<0.05. Both terms are required — the FDR term is what kills the + multiplicity trap (FP-09/FP-10). +2. **RANK** (Phase 2/4): continuous combined score = promotion_score + + anomaly_priority; feeds the bounded top-K human throttle. +3. **CONFIRM** (Phase 2): domain-specific claim bar (5-sigma / FDR+replication / + proof-check). Only CONFIRM-tier candidates are ever claim-eligible. + +The null layer (Phase 1) is mandatory: a candidate with no calibratable null is +tiered UNCALIBRATED and surfaced, never silently passed. Reserved paradigm- +breaker slots (G-05) are guaranteed representation in the shortlist. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from .fdr import benjamini_hochberg, sigma_to_pvalue +from .nulls import InstrumentSystematic +from .promotion import ( + CalibrationStatus, + GateInputs, + LedgerStatus, + anomaly_priority, + calibration_status, + ledger_status, + promotion_score, + reserved_slot_eligible, + surprise_sigma, +) +from .thresholds import Domain, ThresholdPolicy, Tier, confirm_decision + + +@dataclass(frozen=True) +class GateOutcome: + """Per-candidate result after Phase 0/1/2 over the whole family.""" + + candidate_id: str + domain: str + tier: Tier + entered: bool # passed ENTRY (shortlist admission) + rank_score: float # continuous RANK score (promotion + anomaly) + promotion_score: float + anomaly_priority: float + calibration_status: CalibrationStatus + ledger_status: LedgerStatus + sigma_under_null: float | None + fdr_qvalue: float | None + fdr_rejected: bool + surprise_sigma: float + reserved_slot: bool + instrument_systematic_flag: bool # FP-04 surfaced + confirm_reasons: tuple[str, ...] = field(default_factory=tuple) + null_provenance: dict[str, object] = field(default_factory=dict) + + def to_dict(self) -> dict[str, object]: + return { + "candidate_id": self.candidate_id, + "domain": self.domain, + "tier": self.tier.value, + "entered": self.entered, + "rank_score": self.rank_score, + "promotion_score": self.promotion_score, + "anomaly_priority": self.anomaly_priority, + "calibration_status": self.calibration_status.value, + "ledger_status": self.ledger_status.value, + "sigma_under_null": self.sigma_under_null, + "fdr_qvalue": self.fdr_qvalue, + "fdr_rejected": self.fdr_rejected, + "surprise_sigma": self.surprise_sigma, + "reserved_slot": self.reserved_slot, + "instrument_systematic_flag": self.instrument_systematic_flag, + "confirm_reasons": list(self.confirm_reasons), + "null_provenance": self.null_provenance, + } + + +@dataclass(frozen=True) +class FamilyResult: + """The whole-family outcome plus the bounded top-K shortlist.""" + + outcomes: tuple[GateOutcome, ...] + shortlist: tuple[GateOutcome, ...] # bounded top-K (Phase 4 throttle) + policy_hash: str + run_id: str + + @property + def entered_count(self) -> int: + return sum(1 for o in self.outcomes if o.entered) + + def to_dict(self) -> dict[str, object]: + return { + "run_id": self.run_id, + "policy_hash": self.policy_hash, + "scientific_discoveries_claimed": 0, # by construction + "entered_count": self.entered_count, + "outcomes": [o.to_dict() for o in self.outcomes], + "shortlist": [o.candidate_id for o in self.shortlist], + } + + +def _domain(value: str) -> Domain: + try: + return Domain(value) + except ValueError: + return Domain.GENERIC + + +def evaluate_family( + candidates: list[GateInputs], + *, + policy: ThresholdPolicy | None = None, + run_id: str = "run", + orthogonal_by_id: dict[str, bool] | None = None, + proof_checked_by_id: dict[str, bool] | None = None, +) -> FamilyResult: + """Run the full family through Phase 0/1/2 and build the bounded shortlist. + + ``orthogonal_by_id`` / ``proof_checked_by_id`` let the caller supply + confirmation evidence collected after first proposal (the follow-up track); + they default to the per-candidate fields on :class:`GateInputs`. + """ + policy = policy or ThresholdPolicy() + orthogonal_by_id = orthogonal_by_id or {} + proof_checked_by_id = proof_checked_by_id or {} + policy_hash = policy.hash_commit(run_id) + + # Family-wide FDR across the members that have a null sigma (Phase 1/2). + pvals: list[float] = [] + for c in candidates: + sig = c.null.sigma_under_null if c.null.calibratable else None + pvals.append(sigma_to_pvalue(sig) if sig is not None else 1.0) + fdr = benjamini_hochberg(pvals, policy.entry_fdr_q) if candidates else [] + + outcomes: list[GateOutcome] = [] + for i, c in enumerate(candidates): + sig = c.null.sigma_under_null if c.null.calibratable else None + cal = calibration_status(c) + pscore = promotion_score(c) + apriority = anomaly_priority(c, policy=policy) + rank = round(min(pscore + apriority, 1.0), 6) + f = fdr[i] if fdr else None + fdr_rej = bool(f.rejected) if f else False + qval = f.qvalue if f else None + + orthogonal = orthogonal_by_id.get( + c.candidate_id, c.orthogonal_confirmation and c.has_replication + ) + + # FP-06 conservation guard at the TIER level: for a conservation-law + # violation the correct null is measurement error, so until reproducible + # orthogonal confirmation exists the "significance" is not established — + # it is surfaced (UNCALIBRATED), never admitted as a clean signal. + conservation_block = c.violates_conservation_law and not orthogonal + + # ENTRY (Phase 2): 3-sigma-equiv AND family-wide FDR survival AND a + # calibratable null (G-06: L2-holdout is a prerequisite, so an + # UNCALIBRATED candidate is surfaced, never admitted as a clean signal). + meets_sigma = sig is not None and sig >= policy.entry_sigma + entered = bool( + meets_sigma + and fdr_rej + and c.provenance_ok + and cal == CalibrationStatus.CALIBRATED + and not conservation_block + ) + + # CONFIRM (Phase 2): domain-specific claim bar. + conf = confirm_decision( + _domain(c.domain), + sigma=sig, + fdr_rejected=fdr_rej, + orthogonal_replication=orthogonal, + proof_checked=proof_checked_by_id.get(c.candidate_id, False), + policy=policy, + ) + reasons = conf.reasons + if conservation_block: + reasons = (*reasons, "FP-06: conservation-law violation needs orthogonal confirmation") + + # Tiering: UNCALIBRATED surfaces even if it cleared sigma (R6/FP-04/FP-06). + if cal == CalibrationStatus.UNCALIBRATED or conservation_block: + tier = Tier.UNCALIBRATED + elif conf.confirmed and entered: + tier = Tier.CONFIRM + elif entered: + tier = Tier.ENTRY + else: + tier = Tier.UNCALIBRATED + + outcomes.append( + GateOutcome( + candidate_id=c.candidate_id, + domain=c.domain, + tier=tier, + entered=entered, + rank_score=rank, + promotion_score=pscore, + anomaly_priority=apriority, + calibration_status=cal, + ledger_status=ledger_status(c), + sigma_under_null=sig, + fdr_qvalue=qval, + fdr_rejected=fdr_rej, + surprise_sigma=round(surprise_sigma(c), 6), + reserved_slot=reserved_slot_eligible(c, policy=policy), + instrument_systematic_flag=( + c.null.instrument_systematic == InstrumentSystematic.NOT_EXCLUDED + ), + confirm_reasons=reasons, + null_provenance=c.null.to_dict(), + ) + ) + + shortlist = _build_shortlist(outcomes, policy.top_k) + return FamilyResult( + outcomes=tuple(outcomes), + shortlist=tuple(shortlist), + policy_hash=policy_hash, + run_id=run_id, + ) + + +def _build_shortlist(outcomes: list[GateOutcome], top_k: int) -> list[GateOutcome]: + """Bounded top-K by rank, with >=guaranteed reserved paradigm-breaker slots. + + Reserved-slot-eligible candidates get guaranteed representation so the sieve + can never rank paradigm-breakers to zero (recalibration Part 4), while the + overall list stays bounded to protect human review time. + """ + ranked = sorted( + (o for o in outcomes if o.entered or o.reserved_slot), + key=lambda o: o.rank_score, + reverse=True, + ) + reserved = [o for o in ranked if o.reserved_slot] + non_reserved = [o for o in ranked if not o.reserved_slot] + shortlist: list[GateOutcome] = [] + # Guarantee up to 2 reserved slots first (Part 4: ">=2 reserved"). + for o in reserved[:2]: + shortlist.append(o) + for o in non_reserved: + if len(shortlist) >= top_k: + break + shortlist.append(o) + # Backfill any remaining reserved if room. + for o in reserved[2:]: + if len(shortlist) >= top_k: + break + shortlist.append(o) + return shortlist[:top_k] diff --git a/src/sapiens/gates/promotion.py b/src/sapiens/gates/promotion.py new file mode 100644 index 0000000..93f6ae3 --- /dev/null +++ b/src/sapiens/gates/promotion.py @@ -0,0 +1,196 @@ +"""Phase 0 core — recalibrated promotion scoring with the 4 gaming seams welded. + +Implements the recalibrated promotion function (GATES-RECALIBRATION Part 3) with +the four Phase-0 anti-gaming patches and the FP-06 conservation-law guard baked +in so they cannot be bypassed: + +* **G-03** — ``anomaly_priority`` is conditioned on *literature-measured + surprise*, not the mere absence of a mechanism. A trivial unexplained number + earns no boost. +* **G-05** — a reserved paradigm-breaker slot requires + ``promotion_score >= 0.30`` **and** the G-03 surprise condition. A barely-L1 + candidate can no longer squat a reserved slot. +* **G-06** — L2-holdout-passed is a *prerequisite* for CALIBRATED status + (holdout N/A allowed only for deductive / single-observation classes). The + additive 0.50 term only *ranks within* the calibrated set; it is never the + admission key. +* **FP-06** — "no known mechanism" boosts, but "violates a conservation law" + means the correct null is *measurement error* and demands reproducible + orthogonal confirmation before **any** boost. + +Pure stdlib. Score weights follow the recalibration summary; they are a +documented heuristic ranking aid, not a probability. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum + +from .nulls import NullProvenance +from .surprise import LiteratureExpectation, is_surprising, measured_surprise +from .thresholds import ThresholdPolicy + + +class LedgerStatus(str, Enum): + """R1 four-way mechanism x replication ledger (never a kill dimension).""" + + EXPLAINED_CONFIRMED = "explained-confirmed" # mechanism + replication (gold) + UNEXPLAINED_CONFIRMED = "unexplained-confirmed" # the paradigm-breaker signature + EXPLAINED_UNCONFIRMED = "explained-unconfirmed" # follow-up queue + UNEXPLAINED_UNCONFIRMED = "unexplained-unconfirmed" # low initial rank + + +class CalibrationStatus(str, Enum): + CALIBRATED = "calibrated" + UNCALIBRATED = "uncalibrated" # surfaced, never a silent pass (R6) + + +# Recalibrated promotion weights (GATES-RECALIBRATION "promotion function"). +W_PROVENANCE = 0.25 # S1, hard prereq (also gated separately) +W_NULL_SIGMA = 0.30 # S2, continuous, floor 3-sigma to enter +W_METHOD = 0.20 # G5, analysis soundness, NOT consensus +W_REPLICATION = 0.15 # S5, 0 at t0, grows with follow-up +W_ORTHOGONAL = 0.10 # S3, 0 at t0, grows with follow-up + +_SIGMA_SATURATE = 6.0 # sigma at which the null term saturates to 1.0 + + +@dataclass(frozen=True) +class GateInputs: + """Everything a candidate presents to the Phase-0/1/2 gates. + + Deliberately domain-neutral: the adapters translate their measurements into + these fields. Nothing here copies ASTRA/GEODISC/BIODISC/SLATE code. + """ + + candidate_id: str + domain: str # matches thresholds.Domain values where possible + # --- evidence hygiene (hard prereqs live in the runtime kernel) --- + provenance_ok: bool # S1 sky-lock / signal-in-raw-data + method_integrity: float # G5 analysis soundness in [0,1] (NOT consensus) + # --- null layer (Phase 1) --- + null: NullProvenance + # --- maturity dimensions (rank, never admission) --- + has_mechanism: bool + has_replication: bool + orthogonal_confirmation: bool + holdout_passed: bool | None # L2 gate; None => not-applicable class + consensus_conflict: bool = False + # --- G-03 surprise inputs --- + observed_value: float | None = None + literature_expectation: LiteratureExpectation | None = None + # --- FP-06 guard --- + violates_conservation_law: bool = False + # freeform, for the dossier only + notes: str = "" + tags: tuple[str, ...] = field(default_factory=tuple) + + @property + def deductive_or_single_obs(self) -> bool: + """Classes for which L2 holdout is a category error (R7).""" + return self.holdout_passed is None + + +def _null_sigma_term(inputs: GateInputs) -> float: + """Continuous [0,1] term from the significance measured against the null.""" + if not inputs.null.calibratable or inputs.null.sigma_under_null is None: + return 0.0 + return min(max(inputs.null.sigma_under_null, 0.0) / _SIGMA_SATURATE, 1.0) + + +def ledger_status(inputs: GateInputs) -> LedgerStatus: + """The R1 four-way ledger cell for this candidate.""" + if inputs.has_mechanism and inputs.has_replication: + return LedgerStatus.EXPLAINED_CONFIRMED + if inputs.has_replication: + return LedgerStatus.UNEXPLAINED_CONFIRMED + if inputs.has_mechanism: + return LedgerStatus.EXPLAINED_UNCONFIRMED + return LedgerStatus.UNEXPLAINED_UNCONFIRMED + + +def promotion_score(inputs: GateInputs) -> float: + """Recalibrated promotion score in [0,1] (rank aid, not a probability). + + Weighted, additive; maturity dimensions contribute but never subtract, and + mechanism-absence is *not* penalised here (it is handled as an + anomaly-priority boost, R1). Provenance is a weighted term here AND a hard + prereq enforced by the runtime kernel. + """ + score = 0.0 + score += W_PROVENANCE * (1.0 if inputs.provenance_ok else 0.0) + score += W_NULL_SIGMA * _null_sigma_term(inputs) + score += W_METHOD * min(max(inputs.method_integrity, 0.0), 1.0) + score += W_REPLICATION * (1.0 if inputs.has_replication else 0.0) + score += W_ORTHOGONAL * (1.0 if inputs.orthogonal_confirmation else 0.0) + return round(min(score, 1.0), 6) + + +def anomaly_priority(inputs: GateInputs, *, policy: ThresholdPolicy) -> float: + """Inverted mechanism signal (R1) — but gated by G-03 and FP-06. + + * G-03: mechanism-absence contributes only when the candidate is + *measurably surprising* against a cited expectation; an unexplained-but- + unsurprising number earns nothing. + * FP-06: a conservation-law violation means the correct null is measurement + error — no boost at all until orthogonal replication confirms it. + """ + # FP-06 conservation-law guard: measurement error is the correct null. + if inputs.violates_conservation_law and not ( + inputs.has_replication and inputs.orthogonal_confirmation + ): + return 0.0 + priority = 0.0 + surprising = is_surprising( + inputs.observed_value if inputs.observed_value is not None else 0.0, + inputs.literature_expectation, + floor_sigma=policy.surprise_floor_sigma, + ) + # G-03: mechanism-absence boost ONLY when measurably surprising. + if not inputs.has_mechanism and surprising: + priority += 0.10 + # Consensus-conflict is a novelty signal, never a demerit (R2). + if inputs.consensus_conflict: + priority += 0.05 + return round(min(priority, 1.0), 6) + + +def calibration_status(inputs: GateInputs) -> CalibrationStatus: + """G-06 — CALIBRATED requires L2-holdout-passed (or an N/A class) + a + calibratable null. Otherwise UNCALIBRATED (surfaced, never a silent pass). + """ + if not inputs.null.calibratable: + return CalibrationStatus.UNCALIBRATED + # holdout is a prerequisite; None means the deductive/single-obs carve-out. + if inputs.holdout_passed is False: + return CalibrationStatus.UNCALIBRATED + return CalibrationStatus.CALIBRATED + + +def reserved_slot_eligible(inputs: GateInputs, *, policy: ThresholdPolicy) -> bool: + """G-05 — a reserved paradigm-breaker slot requires BOTH a real promotion + score (``>= reserved_slot_min_promotion``) AND the G-03 surprise condition, + and the UNEXPLAINED_CONFIRMED signature. Squatting is structurally blocked. + """ + if ledger_status(inputs) != LedgerStatus.UNEXPLAINED_CONFIRMED: + return False + if promotion_score(inputs) < policy.reserved_slot_min_promotion: + return False + surprising = is_surprising( + inputs.observed_value if inputs.observed_value is not None else 0.0, + inputs.literature_expectation, + floor_sigma=policy.surprise_floor_sigma, + ) + if not surprising: + return False + # FP-06: a conservation-law breaker cannot take a reserved slot unconfirmed. + return not (inputs.violates_conservation_law and not inputs.orthogonal_confirmation) + + +def surprise_sigma(inputs: GateInputs) -> float: + """Convenience: measured surprise in sigma (0.0 if no cited expectation).""" + return measured_surprise( + inputs.observed_value if inputs.observed_value is not None else 0.0, + inputs.literature_expectation, + ) diff --git a/src/sapiens/gates/surprise.py b/src/sapiens/gates/surprise.py new file mode 100644 index 0000000..18fbc8c --- /dev/null +++ b/src/sapiens/gates/surprise.py @@ -0,0 +1,112 @@ +"""G-03 (literature-measured surprise) and G-07 (robust full-dataset baseline). + +Two anti-gaming primitives from HOW-TO-PROCEED Phase 0: + +* **G-07 — baseline gaming.** The MAD/sigma baseline must be computed on the + **full dataset** with a robust estimator over *all* points — never a + pre-filtered "clean" subset that excludes the anomaly and thereby inflates + the apparent significance. :func:`robust_baseline` refuses to drop points and + records the exact point count it saw, so a gamed "baseline on 3 hand-picked + points" is structurally impossible through this API. + +* **G-03 — free anomaly boost.** An anomaly earns priority only when it + *measurably contradicts a literature expectation*, not merely because no + explanation is attached. A trivial unexplained number has zero measured + surprise and therefore earns **no** boost; only a result whose deviation from + a cited expectation exceeds a floor counts as "surprising". + +Pure stdlib (``math``, ``statistics``). +""" + +from __future__ import annotations + +import math +import statistics +from dataclasses import dataclass + +# MAD -> sigma consistency constant for a normal distribution. +_MAD_TO_SIGMA = 1.4826 + + +@dataclass(frozen=True) +class RobustBaseline: + """A robust location/scale estimate computed over the FULL dataset (G-07).""" + + n_points: int # exact number of points used; the anti-gaming receipt + center: float # median (robust location) + scale: float # MAD * 1.4826 (robust sigma), 0.0 only if truly degenerate + used_all_points: bool = True # this API never pre-filters + + def sigma_of(self, value: float) -> float: + """Deviation of ``value`` from the baseline, in robust-sigma units.""" + if self.scale <= 0.0: + return 0.0 + return abs(float(value) - self.center) / self.scale + + +def robust_baseline(values: list[float]) -> RobustBaseline: + """Median + MAD-sigma over **all** ``values`` (G-07: no clean-subset gaming). + + Raises on an empty dataset — a baseline requires data. The estimator is + deliberately robust (median/MAD) so a single extreme anomaly cannot inflate + the scale the way sample-std would; and it consumes every point handed to + it, so callers cannot smuggle in a pre-filtered subset without it showing up + as a smaller ``n_points`` than the source dataset. + """ + vals = [float(v) for v in values] + if not vals: + raise ValueError("robust_baseline requires a non-empty dataset (G-07)") + center = statistics.median(vals) + mad = statistics.median([abs(v - center) for v in vals]) + return RobustBaseline(n_points=len(vals), center=center, scale=mad * _MAD_TO_SIGMA) + + +@dataclass(frozen=True) +class LiteratureExpectation: + """A cited prior expectation a candidate is measured against (G-03). + + ``expected`` and ``expected_sigma`` come from the literature/instrument + error budget, not from the candidate's own dataset — that is what makes the + resulting surprise a *measured contradiction of expectation* rather than a + free boost for any unexplained number. + """ + + expected: float + expected_sigma: float + citation: str + + def __post_init__(self) -> None: + if self.expected_sigma <= 0.0: + raise ValueError("expected_sigma must be positive") + if not self.citation: + raise ValueError("a literature expectation must carry a citation (G-03)") + + +def measured_surprise(observed: float, expectation: LiteratureExpectation | None) -> float: + """Surprise in sigma: how far ``observed`` sits from a *cited* expectation. + + Returns 0.0 when no literature expectation is supplied — the key G-03 + property: **absence of an expectation is not surprise.** A number nobody has + a prediction for earns no anomaly boost; only a measurable contradiction of + a cited prior does. + """ + if expectation is None: + return 0.0 + return abs(float(observed) - expectation.expected) / expectation.expected_sigma + + +def is_surprising( + observed: float, + expectation: LiteratureExpectation | None, + *, + floor_sigma: float = 3.0, +) -> bool: + """True only when the measured surprise clears ``floor_sigma`` (G-03). + + This is the gate the anomaly-priority boost is conditioned on: mere absence + of a mechanism/explanation never satisfies it; a cited, measured + contradiction does. + """ + return math.isfinite(measured_surprise(observed, expectation)) and ( + measured_surprise(observed, expectation) >= floor_sigma + ) diff --git a/src/sapiens/gates/thresholds.py b/src/sapiens/gates/thresholds.py new file mode 100644 index 0000000..d86e9a8 --- /dev/null +++ b/src/sapiens/gates/thresholds.py @@ -0,0 +1,158 @@ +"""Phase 2 — decoupled threshold architecture + cheap boundary fixes. + +Retires the single ``max_pvalue = 0.05`` entry wall and splits three thresholds +that were being conflated: + +============ ================================================================ +Stage Threshold +============ ================================================================ +ENTRY 3-sigma-equiv AND survives BH-FDR q<0.05 across the family +RANK continuous sigma / effect (no cliff; feeds the top-K throttle) +CONFIRM 5-sigma (physics) / FDR<0.05 + orthogonal replication (bio) / + formal proof-check (math) +============ ================================================================ + +One-liner: *"3-sigma + FDR to enter, 5-sigma or replication to claim; 0.05 per +test alone is a multiplicity trap."* + +Boundary fixes folded in: + +* **B-02** adaptive ``ci_floor`` — a margin above the majority-class base rate, + never a fixed 0.70. +* **B-03** hash-commit domain thresholds *before* seeing results. +* **B-05** continuous degree-of-calibration replacing the hard 0.50 quota wall. + +Pure stdlib (``hashlib``, ``json``, ``math``). +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict, dataclass, field +from enum import Enum + + +class Domain(str, Enum): + """Confirmation regime differs by domain (Phase 2 CONFIRM column).""" + + PHYSICS = "physics" # 5-sigma + BIO = "bio" # FDR<0.05 + orthogonal replication + MATH = "math" # formal proof-check + GENERIC = "generic" # default: FDR + orthogonal confirmation + + +@dataclass(frozen=True) +class ThresholdPolicy: + """The full, hash-committable threshold set for one pipeline run (B-03).""" + + entry_sigma: float = 3.0 # ENTRY floor (R5): 3-sigma admits early positives + entry_fdr_q: float = 0.05 # ENTRY also requires family-wide BH-FDR survival + confirm_sigma_physics: float = 5.0 # CONFIRM bar for physics + confirm_fdr_q_bio: float = 0.05 # CONFIRM bar for bio (plus replication) + reserved_slot_min_promotion: float = 0.30 # G-05: reserved-slot prerequisite + surprise_floor_sigma: float = 3.0 # G-03: surprise floor for anomaly boost + top_k: int = 10 # Phase 4 bounded human load + ci_floor_margin: float = 0.15 # B-02: margin above majority-class base rate + devils_advocate_r: float = 0.90 # B-06: r>=this triggers a permutation test + version: str = "gate-hardening-v1" + + def hash_commit(self, run_id: str) -> str: + """SHA-256 over the canonical policy + run id (B-03). + + Call this **before** the run sees any candidate results and record the + digest; it makes post-hoc threshold tuning tamper-evident. + """ + payload = json.dumps( + {"run_id": run_id, "policy": asdict(self)}, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def adaptive_ci_floor(majority_base_rate: float, margin: float = 0.15) -> float: + """B-02: ci_floor = majority-class base rate + margin (never a fixed 0.70). + + On a 95%-majority-class problem a fixed 0.70 floor is *below chance* and + admits noise; on a balanced problem 0.70 may be needlessly harsh. Anchoring + the floor to the base rate plus a fixed margin fixes both. Clamped to + (base_rate, 1.0). + """ + base = min(max(float(majority_base_rate), 0.0), 1.0) + return min(1.0 - 1e-9, max(base + float(margin), base)) + + +def degree_of_calibration( + known_good_covered: int, + known_bad_covered: int, + *, + target_good: int = 5, + target_bad: int = 5, +) -> float: + """B-05: continuous [0,1] calibration coverage, replacing a hard 0.50 wall. + + A hard quota (">=50% of strata calibrated -> CALIBRATED, else refuse") + creates a cliff exactly where borderline candidates live. This returns a + smooth coverage fraction so ranking degrades gracefully instead of flipping + at a wall. + """ + good = min(known_good_covered / target_good, 1.0) if target_good else 1.0 + bad = min(known_bad_covered / target_bad, 1.0) if target_bad else 1.0 + return round(0.5 * (good + bad), 6) + + +class Tier(str, Enum): + """Where a candidate lands after the decoupled thresholds are applied.""" + + UNCALIBRATED = "uncalibrated" # surfaced, never silently passed (R6) + ENTRY = "entry" # on the shortlist; not claimable + CONFIRM = "confirm" # meets the claim bar; human co-sign still required + + +@dataclass(frozen=True) +class ConfirmDecision: + """Outcome of the domain-specific CONFIRM check.""" + + domain: Domain + confirmed: bool + reasons: tuple[str, ...] = field(default_factory=tuple) + + +def confirm_decision( + domain: Domain, + *, + sigma: float | None, + fdr_rejected: bool, + orthogonal_replication: bool, + proof_checked: bool, + policy: ThresholdPolicy, +) -> ConfirmDecision: + """Apply the domain-appropriate CONFIRM bar (the *claim* threshold). + + This is where the old 5-sigma rigor belongs — as a claim bar, not a + candidacy bar. + """ + reasons: list[str] = [] + if domain == Domain.PHYSICS: + ok = sigma is not None and sigma >= policy.confirm_sigma_physics + if not ok: + reasons.append( + f"physics CONFIRM needs sigma>={policy.confirm_sigma_physics} " + f"(got {sigma})" + ) + elif domain == Domain.BIO: + ok = fdr_rejected and orthogonal_replication + if not fdr_rejected: + reasons.append("bio CONFIRM needs family FDR rejection") + if not orthogonal_replication: + reasons.append("bio CONFIRM needs orthogonal replication") + elif domain == Domain.MATH: + ok = proof_checked + if not ok: + reasons.append("math CONFIRM needs a passing formal proof-check") + else: # GENERIC + ok = fdr_rejected and orthogonal_replication + if not ok: + reasons.append("generic CONFIRM needs FDR rejection + orthogonal confirmation") + return ConfirmDecision(domain=domain, confirmed=ok, reasons=tuple(reasons)) diff --git a/tests/test_gates_nulls_thresholds.py b/tests/test_gates_nulls_thresholds.py new file mode 100644 index 0000000..d312ec9 --- /dev/null +++ b/tests/test_gates_nulls_thresholds.py @@ -0,0 +1,108 @@ +"""Phase 1 null-provenance records (+FP-04) and Phase 2 thresholds (+B-02/03/05/06).""" + +import pytest + +from sapiens.gates.devils_advocate import devils_advocate_permutation, pearson_r +from sapiens.gates.nulls import ( + InstrumentSystematic, + NullKind, + NullProvenance, + no_null, +) +from sapiens.gates.thresholds import ( + Domain, + ThresholdPolicy, + adaptive_ci_floor, + confirm_decision, + degree_of_calibration, +) + + +def test_null_provenance_records_external_fetch(): + n = NullProvenance( + kind=NullKind.CORRECT, + description="Planck 353 GHz dust map", + external_data_required=True, + external_data_fetched=False, + sigma_under_null=5.0, + ) + assert not n.data_complete # required data never fetched + assert not n.calibratable # so it cannot calibrate a claim + assert "external_data_fetched" in n.to_dict() + + +def test_fp04_instrument_systematic_blocks_calibration(): + n = NullProvenance( + kind=NullKind.CORRECT, + description="timing series", + external_data_required=False, + external_data_fetched=True, + sigma_under_null=6.0, + instrument_systematic=InstrumentSystematic.NOT_EXCLUDED, + ) + assert not n.calibratable # a 6-sigma with an un-excluded systematic is NOT clean + + +def test_no_null_is_not_calibratable(): + n = no_null() + assert n.kind == NullKind.NONE + assert not n.is_constructed + assert not n.calibratable + + +def test_b02_adaptive_ci_floor_beats_fixed_070(): + # 95%-majority class: a fixed 0.70 is below chance; adaptive floor is above it. + assert adaptive_ci_floor(0.95, margin=0.03) > 0.95 + assert adaptive_ci_floor(0.5, margin=0.15) == pytest.approx(0.65) + assert adaptive_ci_floor(0.99, margin=0.10) < 1.0 # clamped + + +def test_b05_degree_of_calibration_is_continuous(): + assert degree_of_calibration(0, 0) == 0.0 + assert degree_of_calibration(5, 5) == 1.0 + mid = degree_of_calibration(2, 3) + assert 0.0 < mid < 1.0 # no hard 0.50 cliff + + +def test_b03_hash_commit_is_deterministic_and_binds_run(): + p = ThresholdPolicy() + assert p.hash_commit("run-a") == p.hash_commit("run-a") + assert p.hash_commit("run-a") != p.hash_commit("run-b") + assert len(p.hash_commit("x")) == 64 + + +def test_confirm_bars_are_domain_specific(): + p = ThresholdPolicy() + # physics: 5-sigma. + assert confirm_decision(Domain.PHYSICS, sigma=5.0, fdr_rejected=True, + orthogonal_replication=False, proof_checked=False, + policy=p).confirmed + assert not confirm_decision(Domain.PHYSICS, sigma=4.0, fdr_rejected=True, + orthogonal_replication=True, proof_checked=True, + policy=p).confirmed + # bio: FDR + orthogonal replication. + assert confirm_decision(Domain.BIO, sigma=3.0, fdr_rejected=True, + orthogonal_replication=True, proof_checked=False, + policy=p).confirmed + assert not confirm_decision(Domain.BIO, sigma=9.0, fdr_rejected=True, + orthogonal_replication=False, proof_checked=False, + policy=p).confirmed + # math: formal proof-check. + assert confirm_decision(Domain.MATH, sigma=None, fdr_rejected=False, + orthogonal_replication=False, proof_checked=True, + policy=p).confirmed + + +def test_b06_devils_advocate_permutation(): + # Signal perfectly tracks a confound -> guard fires, and it does NOT pass. + confound = [float(i) for i in range(20)] + signal = [2 * x + 1 for x in confound] # r == 1.0 + assert pearson_r(signal, confound) == pytest.approx(1.0) + res = devils_advocate_permutation(signal, confound, seed=1, n_permutations=500) + assert res.triggered + assert not res.passed # confound fully explains it + # Weak correlation -> guard never fires, passes trivially. + weak = devils_advocate_permutation( + [1.0, 2.0, 1.5, 3.0, 2.5], [9.0, 1.0, 5.0, 2.0, 8.0], seed=1 + ) + assert not weak.triggered and weak.passed diff --git a/tests/test_gates_phases345.py b/tests/test_gates_phases345.py new file mode 100644 index 0000000..9af2157 --- /dev/null +++ b/tests/test_gates_phases345.py @@ -0,0 +1,133 @@ +"""Phase 3 (blind harness), Phase 4 (human-gate dossier), Phase 5 (criteria), +plus the clean-room invariant for the gates subpackage.""" + +import ast +from pathlib import Path + +from sapiens.gates.blind import ( + Verdict, + blind_run, + grade, + seal, + strip_to_blind, +) +from sapiens.gates.corpus import decoys, full_corpus, historical_positives +from sapiens.gates.criteria import run_success_criteria +from sapiens.gates.dossier import ( + OverrideLog, + autonomous_claim_eligible, + build_dossier, +) +from sapiens.gates.pipeline import evaluate_family +from sapiens.gates.thresholds import Tier + +ROOT = Path(__file__).resolve().parents[1] + + +# --- Phase 3: blind re-run harness --------------------------------------- # +def test_blind_strip_removes_key_fields(): + item = historical_positives()[0] + blind = strip_to_blind(item) + # The stripped object is just the model inputs — no ground_truth/verdict/how_*. + assert not hasattr(blind, "ground_truth") + assert not hasattr(blind, "expected_verdict") + assert not hasattr(blind, "how_old_gate_would_kill") + + +def test_blind_custody_commitment_and_grade(): + corpus = full_corpus() + sealed = seal(corpus, seed=3) + assert sealed.key.verify_commitment() + verdicts = blind_run(sealed, run_id="blind-test") + report = grade(verdicts, sealed.key) + # Real null layer, answers not inline: kill/recovery targets met, low abstention. + assert report.kill_rate >= 0.8 + assert report.recovery_rate >= 0.8 + assert report.abstention_rate < 0.10 + assert report.leaks == () + + +def test_blind_scorer_sees_no_labels(): + # The blind runner only receives GateInputs; assert none carry a label attr. + sealed = seal(full_corpus(), seed=1) + for gi in sealed.blind_inputs: + assert not hasattr(gi, "label") + + +def test_tampered_custody_key_is_rejected(): + sealed = seal(decoys(), seed=1) + bad = sealed.key.__class__(labels={"x": "positive"}, commitment="deadbeef") + verdicts = blind_run(sealed) + try: + grade(verdicts, bad) + except ValueError: + return + raise AssertionError("tampered custody key should have been rejected") + + +# --- Phase 4: human-in-loop gate ----------------------------------------- # +def test_dossier_forces_disconfirming_explanation(): + fam = evaluate_family([it.inputs for it in historical_positives()], run_id="d") + for o in fam.outcomes: + d = build_dossier(o) + assert d.strongest_disconfirming_explanation # never empty + assert "null_used" in d.to_dict() + + +def test_only_confirm_tier_is_claim_eligible(): + fam = evaluate_family([it.inputs for it in full_corpus()], run_id="d") + for o in fam.outcomes: + eligible = autonomous_claim_eligible(o) + assert eligible == (o.tier == Tier.CONFIRM) + # Nothing below CONFIRM is ever autonomously claimable. + if o.tier in (Tier.ENTRY, Tier.UNCALIBRATED): + assert not eligible + + +def test_override_log_feeds_labeled_data(): + fam = evaluate_family([it.inputs for it in historical_positives()], run_id="d") + log = OverrideLog() + log.record(fam.outcomes[0], human_verdict="pursue", rationale="looks real", + timestamp="2026-07-24T00:00:00Z") + log.record(fam.outcomes[1], human_verdict="reject", rationale="artifact", + timestamp="2026-07-24T00:01:00Z") + labels = log.as_training_labels() + assert len(labels) == 2 + assert all(v in ("pursue", "reject", "defer") for _, v in labels) + assert 0.0 <= log.disagreement_rate() <= 1.0 + + +# --- Phase 5: success-criteria suite ------------------------------------- # +def test_success_criteria_all_pass(): + suite = run_success_criteria() + failing = [c.name for c in suite.criteria if not c.passed] + assert not failing, f"failing criteria: {failing}" + assert suite.all_passed + assert suite.to_dict()["scientific_discoveries_claimed"] == 0 + + +def test_verdict_vocabulary_is_total(): + sealed = seal(full_corpus(), seed=2) + verdicts = blind_run(sealed) + assert all(isinstance(v, Verdict) for v in verdicts.values()) + + +# --- clean-room invariant for the gates subpackage ----------------------- # +def test_gates_subpackage_is_clean_room(): + forbidden_imports = ("sapiens.adapters",) + forbidden_names = ("astra", "geodisc", "biodisc", "slate") + for path in (ROOT / "src" / "sapiens" / "gates").glob("*.py"): + tree = ast.parse(path.read_text()) + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + assert not (node.module or "").startswith(forbidden_imports), path + elif isinstance(node, ast.Import): + for alias in node.names: + assert not alias.name.startswith(forbidden_imports), path + # No copied ASTRA-family identifiers leaked into code (docstrings ok). + lowered = "\n".join( + line for line in path.read_text().splitlines() + if not line.strip().startswith("#") + ).lower() + for name in forbidden_names: + assert f"import {name}" not in lowered, (path, name) diff --git a/tests/test_gates_pipeline_gaming.py b/tests/test_gates_pipeline_gaming.py new file mode 100644 index 0000000..24b9a67 --- /dev/null +++ b/tests/test_gates_pipeline_gaming.py @@ -0,0 +1,80 @@ +"""Integration: decoupled ENTRY/RANK/CONFIRM tiers, family FDR, the FP-06 tier +block, and the Phase-0 exit criterion (7 gaming vectors -> 0 leaks).""" + +from sapiens.gates.corpus import gaming_vectors, historical_positives +from sapiens.gates.criteria import gaming_leaked, run_gaming_retest +from sapiens.gates.nulls import NullKind, NullProvenance +from sapiens.gates.pipeline import evaluate_family +from sapiens.gates.promotion import GateInputs +from sapiens.gates.surprise import LiteratureExpectation +from sapiens.gates.thresholds import ThresholdPolicy, Tier + + +def _null(sigma, **kw): + return NullProvenance( + kind=NullKind.CORRECT, + description="null", + external_data_required=kw.get("req", False), + external_data_fetched=kw.get("fetched", True), + sigma_under_null=sigma, + ) + + +def test_entry_requires_sigma_and_family_fdr(): + # A 6-sigma physics candidate in a small clean family enters and confirms. + strong = GateInputs( + candidate_id="strong", domain="physics", provenance_ok=True, + method_integrity=0.9, null=_null(6.0), has_mechanism=True, + has_replication=True, orthogonal_confirmation=True, holdout_passed=None, + observed_value=6.0, literature_expectation=LiteratureExpectation(0.0, 1.0, "l"), + ) + res = evaluate_family([strong], run_id="t") + o = res.outcomes[0] + assert o.entered + assert o.tier == Tier.CONFIRM + assert res.policy_hash # thresholds hash-committed + + +def test_conservation_violation_never_enters_without_orthogonal(): + breaker = GateInputs( + candidate_id="oa", domain="physics", provenance_ok=True, + method_integrity=0.8, null=_null(7.0), has_mechanism=False, + has_replication=False, orthogonal_confirmation=False, holdout_passed=None, + observed_value=42.0, violates_conservation_law=True, + literature_expectation=LiteratureExpectation(0.0, 3.0, "conservation"), + ) + o = evaluate_family([breaker], run_id="t").outcomes[0] + assert not o.entered + assert o.tier == Tier.UNCALIBRATED + assert any("FP-06" in r for r in o.confirm_reasons) + + +def test_shortlist_is_bounded_and_reserves_paradigm_breakers(): + fam = evaluate_family( + [it.inputs for it in historical_positives()], + policy=ThresholdPolicy(top_k=10), + run_id="hist", + ) + assert len(fam.shortlist) <= 10 # bounded human load + # At least one reserved paradigm-breaker slot is represented. + assert any(o.reserved_slot for o in fam.shortlist) + + +def test_phase0_exit_zero_gaming_leaks(): + retest = run_gaming_retest() + assert retest.checked == 7 + assert retest.leaks == () # 0 leaks (was 4 before the patches) + assert retest.clean + + +def test_each_gaming_vector_denied_its_reward(): + # Per-candidate check that each seam denies exactly its illegitimate reward. + vectors = {v.inputs.candidate_id: v for v in gaming_vectors()} + per = [v for v in vectors.values() if v.inputs.candidate_id != "gv-fp09-multiplicity"] + outcomes = { + o.candidate_id: o + for o in evaluate_family([v.inputs for v in per], run_id="g").outcomes + } + for cid, o in outcomes.items(): + reward = vectors[cid].illegitimate_reward + assert not gaming_leaked(vectors[cid], o), f"{cid} leaked {reward}" diff --git a/tests/test_gates_promotion.py b/tests/test_gates_promotion.py new file mode 100644 index 0000000..4e8ed9c --- /dev/null +++ b/tests/test_gates_promotion.py @@ -0,0 +1,134 @@ +"""Phase 0 core: G-03/G-05/G-06 seams + FP-06 guard on the promotion path.""" + +from sapiens.gates.nulls import InstrumentSystematic, NullKind, NullProvenance +from sapiens.gates.promotion import ( + CalibrationStatus, + GateInputs, + LedgerStatus, + anomaly_priority, + calibration_status, + ledger_status, + promotion_score, + reserved_slot_eligible, +) +from sapiens.gates.surprise import LiteratureExpectation +from sapiens.gates.thresholds import ThresholdPolicy + +POLICY = ThresholdPolicy() + + +def _null(sigma, **kw): + return NullProvenance( + kind=kw.get("kind", NullKind.CORRECT), + description=kw.get("description", "null"), + external_data_required=kw.get("req", False), + external_data_fetched=kw.get("fetched", True), + sigma_under_null=sigma, + instrument_systematic=kw.get("inst", InstrumentSystematic.NOT_APPLICABLE), + ) + + +def _inputs(**kw): + base = dict( + candidate_id="c", + domain="physics", + provenance_ok=True, + method_integrity=0.8, + null=_null(5.0), + has_mechanism=False, + has_replication=True, + orthogonal_confirmation=False, + holdout_passed=None, + observed_value=5.0, + literature_expectation=LiteratureExpectation(0.0, 1.0, "lit"), + ) + base.update(kw) + return GateInputs(**base) + + +def test_ledger_four_way(): + assert ledger_status(_inputs(has_mechanism=True, has_replication=True)) == ( + LedgerStatus.EXPLAINED_CONFIRMED + ) + assert ledger_status(_inputs(has_mechanism=False, has_replication=True)) == ( + LedgerStatus.UNEXPLAINED_CONFIRMED + ) + assert ledger_status(_inputs(has_mechanism=False, has_replication=False)) == ( + LedgerStatus.UNEXPLAINED_UNCONFIRMED + ) + + +def test_g03_anomaly_boost_requires_surprise(): + # Unexplained but matches expectation (surprise 0) -> no boost. + unsurprising = _inputs( + has_mechanism=False, observed_value=0.0, + literature_expectation=LiteratureExpectation(0.0, 1.0, "lit"), + ) + assert anomaly_priority(unsurprising, policy=POLICY) == 0.0 + # Unexplained AND measurably surprising -> boost. + surprising = _inputs( + has_mechanism=False, observed_value=8.0, + literature_expectation=LiteratureExpectation(0.0, 1.0, "lit"), + ) + assert anomaly_priority(surprising, policy=POLICY) > 0.0 + + +def test_fp06_conservation_violation_denies_boost(): + breaker = _inputs( + has_mechanism=False, observed_value=42.0, violates_conservation_law=True, + has_replication=False, orthogonal_confirmation=False, + literature_expectation=LiteratureExpectation(0.0, 1.0, "conservation"), + ) + assert anomaly_priority(breaker, policy=POLICY) == 0.0 + # Only orthogonal confirmation + replication unlocks any consideration. + confirmed = _inputs( + has_mechanism=False, observed_value=42.0, violates_conservation_law=True, + has_replication=True, orthogonal_confirmation=True, + literature_expectation=LiteratureExpectation(0.0, 1.0, "conservation"), + ) + assert anomaly_priority(confirmed, policy=POLICY) >= 0.0 # no crash; boost allowed + + +def test_g06_calibrated_requires_holdout(): + # Calibratable null but L2 holdout explicitly failed -> UNCALIBRATED. + assert calibration_status(_inputs(holdout_passed=False)) == ( + CalibrationStatus.UNCALIBRATED + ) + # holdout passed -> CALIBRATED. + assert calibration_status(_inputs(holdout_passed=True)) == ( + CalibrationStatus.CALIBRATED + ) + # None = deductive/single-obs carve-out -> CALIBRATED if null is calibratable. + assert calibration_status(_inputs(holdout_passed=None)) == ( + CalibrationStatus.CALIBRATED + ) + # No calibratable null -> UNCALIBRATED regardless of holdout. + assert calibration_status( + _inputs(null=_null(5.0, req=True, fetched=False)) + ) == CalibrationStatus.UNCALIBRATED + + +def test_g05_reserved_slot_requires_score_and_surprise(): + # Barely-L1 squatter: weak provenance/method -> low promotion score. + squatter = _inputs( + provenance_ok=False, method_integrity=0.05, null=_null(0.5), + has_mechanism=False, has_replication=True, observed_value=100.0, + literature_expectation=LiteratureExpectation(0.0, 1.0, "lit"), + ) + assert promotion_score(squatter) < POLICY.reserved_slot_min_promotion + assert not reserved_slot_eligible(squatter, policy=POLICY) + # Genuine paradigm-breaker: strong score + surprise + UNEXPLAINED_CONFIRMED. + real = _inputs( + provenance_ok=True, method_integrity=0.9, null=_null(8.0), + has_mechanism=False, has_replication=True, observed_value=8.0, + literature_expectation=LiteratureExpectation(0.0, 1.0, "lit"), + ) + assert promotion_score(real) >= POLICY.reserved_slot_min_promotion + assert reserved_slot_eligible(real, policy=POLICY) + + +def test_promotion_score_never_penalises_mechanism_absence(): + with_mech = _inputs(has_mechanism=True) + without = _inputs(has_mechanism=False) + # Mechanism presence is not a promotion-score term (it lives in anomaly track). + assert promotion_score(with_mech) == promotion_score(without) diff --git a/tests/test_gates_stats.py b/tests/test_gates_stats.py new file mode 100644 index 0000000..6c96fc0 --- /dev/null +++ b/tests/test_gates_stats.py @@ -0,0 +1,81 @@ +"""Phase 0/1 statistical primitives: robust baseline (G-07), surprise (G-03), +BH-FDR + sigma conversions (Phase 1/2).""" + +import math + +import pytest + +from sapiens.gates.fdr import benjamini_hochberg, pvalue_to_sigma, sigma_to_pvalue +from sapiens.gates.surprise import ( + LiteratureExpectation, + is_surprising, + measured_surprise, + robust_baseline, +) + + +def test_sigma_pvalue_roundtrip(): + for s in (1.0, 3.0, 5.0, 6.5): + assert pvalue_to_sigma(sigma_to_pvalue(s)) == pytest.approx(s, abs=1e-3) + assert sigma_to_pvalue(3) == pytest.approx(0.0027, abs=1e-4) + assert sigma_to_pvalue(5) == pytest.approx(5.7e-7, rel=0.05) + + +def test_g07_robust_baseline_uses_full_dataset(): + # A noisy full dataset: a 9.5 outlier is only ~1 robust-sigma; a cherry- + # picked clean subset would inflate that to tens of sigma (the G-07 exploit). + full = [-8, -6, -5, -4, -3, 3, 4, 5, 6, 8, 9.5, 0] + b_full = robust_baseline(full) + assert b_full.n_points == len(full) # the anti-gaming receipt + assert b_full.sigma_of(9.5) < 3.0 + clean = [0, 0.1, -0.1, 0.2, -0.2, 0.05] + assert robust_baseline(clean).sigma_of(9.5) > 20.0 + + +def test_robust_baseline_rejects_empty(): + with pytest.raises(ValueError): + robust_baseline([]) + + +def test_g03_surprise_requires_measured_contradiction(): + exp = LiteratureExpectation(expected=10.0, expected_sigma=1.0, citation="lit[x]") + # An unexplained-but-unsurprising number (matches expectation) => 0 surprise. + assert measured_surprise(10.0, exp) == pytest.approx(0.0) + assert not is_surprising(10.0, exp, floor_sigma=3.0) + # A measurable contradiction is surprising. + assert measured_surprise(15.0, exp) == pytest.approx(5.0) + assert is_surprising(15.0, exp, floor_sigma=3.0) + # Absence of a cited expectation is NOT surprise. + assert measured_surprise(1e9, None) == 0.0 + assert not is_surprising(1e9, None) + + +def test_literature_expectation_requires_citation(): + with pytest.raises(ValueError): + LiteratureExpectation(0.0, 1.0, "") + with pytest.raises(ValueError): + LiteratureExpectation(0.0, 0.0, "lit") + + +def test_family_wide_bh_fdr_controls_multiplicity(): + # One genuine 4-sigma signal buried among many null looks. + pvals = [sigma_to_pvalue(4.0)] + [0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.3, 0.55] + res = benjamini_hochberg(pvals, q=0.05) + assert res[0].rejected # the real one survives family FDR + assert sum(r.rejected for r in res) == 1 # the noise does not + # q-values are monotone and in-order with input. + assert all(0.0 <= r.qvalue <= 1.0 for r in res) + + +def test_bh_look_elsewhere_kills_lone_three_sigma(): + # A single per-test 3-sigma among ~1000 pure looks must NOT survive. + pvals = [sigma_to_pvalue(3.1)] + [0.5] * 999 + res = benjamini_hochberg(pvals, q=0.05) + assert not res[0].rejected + + +def test_bh_empty_and_bounds(): + assert benjamini_hochberg([], 0.05) == [] + with pytest.raises(ValueError): + benjamini_hochberg([0.1], 0.0) + assert math.isfinite(sigma_to_pvalue(0.0))