diff --git a/.kiro/specs/learned-decision-engine/requirements.md b/.kiro/specs/learned-decision-engine/requirements.md index 84af32d..345a655 100644 --- a/.kiro/specs/learned-decision-engine/requirements.md +++ b/.kiro/specs/learned-decision-engine/requirements.md @@ -119,6 +119,19 @@ rule hits. 2. Without a trained model available the tools fall back to today's behaviour and say so in the report. 3. Hand-written rules remain as signal producers; changing their enabled/shadow status no longer changes what is reported unless the engine's decision changes. +4. *(Amended 2026-10-03, maintainer decision; refines the BLOCK rule of 7.3 and 7.4. Tested by exp-004.)* BLOCK + precision is demonstrated pooled across families, not per family: + - **Normalisation**, declared before any run: each family's raw score is put on one scale by a per-family Platt + map (`learn.calibrate.fit_map` with its floors) fitted on training folds only. A family the training folds + cannot map does not block on that fold. + - **t_block**: one threshold on the normalised score for all families, chosen on training folds as the lowest + one whose Wilson 95% lower bound on precision, over the training flags of all families together, is >= 0.95. + It is reported by the Wilson 95% lower bound over all held-out flags of all families, before the floor. + - **Family floor**: a family may block only if its own held-out flags have zero errors and number at least F + (F = 10, reasoned in the exp-004 registration); otherwise that family stays ADVISORY. + - **t_quiet**: per family, the highest threshold keeping >= 90% of the paired positives at or above it (chosen + on training folds, its held-out recall reported); findings below it are silent. + - No held-out label moves a threshold or a normalisation map; candidates whose verdict is `unsure` never block. ### Requirement 7: Built for unseen repositories -- a repo safety net @@ -136,7 +149,8 @@ rule hits. ADVISORY, trading precision for recall; both reported with out-of-repository intervals. The decision unit is a candidate in changed code (`pre-push` delta), with whole-repository scans as a secondary mode. 4. Calibration is checked on held-out repositories; where it does not hold, only ADVISORY is offered for that - family. + family. *(Amended 2026-10-03: once exp-004 adopts the pooled gate of 6.4, BLOCK precision is shown by its held-out + bound and family floor, and calibration is reported, not required.)* 5. Any adaptation to a user's own repository (facts reuse, their dismissals) is opt-in, separate from the generalisation model, and never contributes to its reported numbers. diff --git a/plane/experiments/exp-004-pooled-block-gate.yaml b/plane/experiments/exp-004-pooled-block-gate.yaml new file mode 100644 index 0000000..057b89b --- /dev/null +++ b/plane/experiments/exp-004-pooled-block-gate.yaml @@ -0,0 +1,90 @@ +# Experiment exp-004 (learned-decision-engine Req 6.4, amended 2026-10-03 by maintainer decision): the pooled BLOCK +# gate with a family floor. Registered before it runs; `ousast learn experiment register` records this file's sha256, +# and a changed file is a new experiment. The procedure and the decision rule are fixed here; the units are NOT: they +# are frozen at run time (`ousast learn experiment units`) from the memory after the negatives harvest (merged in +# 5b2bd9b, 1133 static examples), committed, and registered again before the run. Until then the registration row +# says `units_status: pending` and the run refuses. +# +# Scoring: kind rule, zero cost by construction -- one scoring pass of the six slice-2 incumbent programs, replayed from +# the response cache behind a zero-ceiling meter; a unit with any uncached response is excluded and counted. The +# harvest changed the memory the programs retrieve from, so most prompts are new: a paid cache fill (separately +# approved, its own ceiling, as for exp-003's supplementary analysis) must precede the run, or the exclusions will be +# most of the units. +# +# Normalisation (declared before the run): per-family Platt scaling, learn.calibrate.fit_map with its existing floors +# (own slope and intercept per family at >= 20 positive and negative repository groups; a pooled slope with a +# per-family intercept at >= 10; below that no map, so the family cannot block on that fold). Fitted on the training +# folds' paired units only. Why Platt and not rank-to-quantile: a quantile map gives every family the same flag rate +# whatever its precision, so one pooled threshold would flag the top slice of a weak family as readily as of a strong +# one. Platt puts each family on the scale of its own training precision, which is what a pooled precision bound needs. +# The map is a normaliser only: exp-003 showed calibration (ECE <= 0.05) does not hold per family, and precision is +# demonstrated here by the bound on held-out flags, not by calibration. +# +# Thresholds, per outer fold j, from the other four folds only (fold j's labels never enter a threshold or a map; +# tests/test_learn_rules.py): +# - the training units are normalised by an inner cross-fit (each training fold k by a map fitted on the training +# folds other than k), so t_block is chosen on out-of-sample scores; +# - t_block: the lowest normalised score at which the training paired units of ALL families together with p >= t +# and a verdict other than `unsure` have a Wilson 95% lower bound of precision >= 0.95; none -> no BLOCK on fold j; +# - t_quiet, per family: the highest normalised score keeping >= 90% of that family's training paired positives at +# or above it; +# - fold j's units, normalised by the map fitted on all its training units: BLOCK when not `unsure` and p >= t_block; +# ADVISORY when p >= t_quiet (or the family has no map or no t_quiet: nothing is silenced blind); silent below. +# +# Family floor F = 10: a family may block only if its own held-out paired flags (all five folds) number >= 10 with zero +# errors; otherwise its flags are reported as ADVISORY. Why 10: zero errors in 10 flags is unlikely for a family that is +# much worse than the bar (a family at 0.75 precision produces 10 clean flags with probability 0.056, at 0.80 with +# 0.107); a larger F is out of reach of the smaller families (16-19 paired positives in exp-003's units, all folds); and +# 10 is exp-003 arm C's training minimum. The floor uses held-out labels to decide which families block; it moves no +# threshold. The primary metric is computed BEFORE the floor, over every family's held-out flags, so dropping a failing +# family cannot raise it. +# +# Power note: the pooled bound needs >= 73 flagged units with no error (110 with one, 142 with two). exp-003's +# instrument check found few positives above the first negative per family on the raw score (paired: 10, 1, 3, 0, 1); +# per-family normalisation keeps each family's order, so the pooled top set can reach 73 only if the harvest's pairs +# separate better. An inconclusive verdict is a plausible outcome and will be reported as such. +id: exp-004-pooled-block-gate +hypothesis: > + One BLOCK threshold pooled across families on a per-family Platt-normalised score, chosen on training folds only as + the lowest one whose Wilson 95% lower bound on flagged precision is >= 0.95, reaches that bound on held-out + repositories' paired units, where per-family thresholds (exp-003) could not for lack of flags; a per-family floor + (>= 10 held-out flags, none wrong) keeps a family that the pooling masks from blocking. +kind: rule +programs: # the slice-2 incumbents, as exp-003 + access_control: 745cb1d76bbb09dc1574d333d2118489be62e39886517894d3699fa565b0e8c2 + deserialization: 3c0c2831be7d78198a62351661c37867363c501cb40e3ed7b0119c155af09654 + injection: b207a0710e0abf053b3b9397b25f9ee1cab4679144865e3a82c1b9be2c7b4982 + output_encoding: 5d768f16724562c381c09949ff3df9cbb11d58cbb7a9c0515d0dc3f858d3818e + path: aea396680a4874dcb6de70e240ca0480c83424debdd3710bc3f1ea6a612661f3 + untrusted_destination: 1242b3973fc89056aa08e00843043d68235df217d8e3ef8a0ea0e0b81c142753 +arms: + A: + label: today's rule -- Platt map cross-fitted per family, BLOCK only when the calibration check holds, BLOCK point nested on the other folds' calibrated p + rule: {name: calibrated_block} + B: + label: pooled gate -- one t_block over all families on a per-family Platt score (training folds, inner cross-fit), t_quiet per family at 90% recall, family floor 10 clean held-out flags + rule: {name: pooled_block_gate, precision: 0.95, recall: 0.9, floor: 10, normalise: platt, select_on: paired} +unit: candidate +pairing: paired # every unit scored once; every arm decides every unit +cluster: repository +order: {randomise: none, seed: 4} +# primary population: paired units (both sides of the candidate present); pooled (every unit) is reported beside it +populations: {primary: paired, secondary: [pooled]} +metric: + primary: + - {name: flagged_precision_lb, target: 0.95} # held-out Wilson 95% LB of flagged precision, all families pooled, arm B, paired, before the floor + secondary: [flagged_precision, coverage, flagged, quiet_recall] # quiet_recall: held-out positives at or above t_quiet / positives +# Decision rule (decide_pooled), on arm B's held-out paired flags of all families together, before the floor: +# REJECT B if their point precision < 0.95; +# else ADOPT B if their Wilson 95% LB >= 0.95 and at least one family passes the floor (>= 10 held-out paired +# flags, zero errors) -- those families block, every other family stays advisory; +# else INCONCLUSIVE (the incumbent rule stays). +# Reported, not deciding: per-family held-out precision and LB, the floor table, quiet_recall per family against 0.90, +# the after-floor totals (what users would get), the repository-cluster bootstrap interval. +decision: {arm: B, target_precision: 0.95, pooled: true} +test: {resamples: 2000} # repository-cluster bootstrap of the held-out flagged precision, beside the Wilson bound +mde: 0.0 # not a paired difference test; see the power note above +budget_usd: {total: 0} # the run itself is zero cost; the cache fill before it is a separate, approved spend +families: [access_control, deserialization, injection, output_encoding, path, untrusted_destination] +folds: {kind: outer, k: 5, seed: 0, compile_fraction: 0.25} # the incumbent evaluation's folds (learn/compile.toml seed) +units_file: plane/experiments/exp-004-pooled-block-gate.units.jsonl # frozen at run time, after the negatives harvest diff --git a/src/openultrasast/learn/evaluate.py b/src/openultrasast/learn/evaluate.py index 12a6621..df2d29d 100644 --- a/src/openultrasast/learn/evaluate.py +++ b/src/openultrasast/learn/evaluate.py @@ -25,7 +25,7 @@ from collections.abc import Callable, Iterable, Mapping, Sequence from dataclasses import dataclass, replace from typing import Any -from .calibrate import Row, check, cross_fit, offered +from .calibrate import Row, check, cross_fit, fit_map, offered from .examples import Example from .folds import Fold from .labels import PER_FAMILY_FLOOR, POOLED_FLOOR @@ -210,6 +210,98 @@ def calibrated_block(preds: Sequence[Prediction], *, seed: int = 0, **floors: in return out, info +# --- the pooled BLOCK gate (Req 6.4, amended 2026-10-03; exp-004) --------------------------------------------------------- + +FAMILY_FLOOR = 10 # a family blocks only with at least this many held-out flags and none of them wrong + + +def normalised(train: Sequence[Prediction], rows: Sequence[Prediction], **floors: int) -> list[float | None]: + """``rows``' raw scores on one scale across families: the per-family Platt map of :func:`.calibrate.fit_map` + (pooled slope with a per-family intercept between the floors) fitted on ``train`` only. ``None`` for a family the + training rows cannot map (below the pooled floor): it cannot block on that fold.""" + cmap = fit_map([Row(r.fold, r.family, r.group, r.label, r.s) for r in train], **floors) + return [cmap.apply(r.family, r.s) for r in rows] + + +def pooled_threshold(rows: Sequence[Prediction], *, precision: float = TARGET_PRECISION) -> float | None: + """The lowest normalised score ``t`` at which the blockable rows of every family together with ``p >= t`` have a + Wilson 95% lower bound of precision >= ``precision``; candidate thresholds are the rows' own ``p``.""" + blockable = [r for r in rows if r.blockable and r.p is not None] + for t in sorted({float(r.p) for r in blockable if r.p is not None}): + chosen = [r for r in blockable if (r.p or 0.0) >= t] + if wilson_lower(sum(r.label for r in chosen), len(chosen)) >= precision: + return t + return None + + +def quiet_threshold(rows: Sequence[Prediction], *, recall: float = TARGET_RECALL) -> float | None: + """The highest normalised score ``t`` that keeps at least ``recall`` of the rows' positives at or above it.""" + positives = sorted((float(r.p) for r in rows if r.label == 1 and r.p is not None), reverse=True) + if not positives: + return None + return positives[math.ceil(recall * len(positives) - 1e-9) - 1] + + +def pooled_block_gate( + preds: Sequence[Prediction], + *, + eligible: Callable[[Prediction], bool] | None = None, + precision: float = TARGET_PRECISION, + recall: float = TARGET_RECALL, + **floors: int, +) -> tuple[list[Prediction], dict[str, dict[str, Any]]]: + """One BLOCK threshold for every family, on a normalised score, chosen per outer fold j on the other folds only. + + The training rows (``eligible`` rows of the other folds) are normalised by an inner cross-fit: each training fold + k by a map fitted on the training folds other than k, so ``t_block`` is chosen on out-of-sample scores. ``t_block`` + is the lowest normalised score whose pooled Wilson 95% lower bound of precision is >= ``precision``; ``t_quiet`` + per family is the highest one keeping >= ``recall`` of that family's training positives. Fold j's rows are then + normalised by the map fitted on all its training rows and decided: ``BLOCK`` when blockable and ``p >= t_block``; + ``ADVISORY`` when ``p >= t_quiet`` (or the family has no map, or no ``t_quiet``: nothing is silenced blind); + ``DROP`` (silent) below ``t_quiet``. Fold j's labels never enter its thresholds or its map.""" + chosen = [r for r in preds if eligible is None or eligible(r)] + families = sorted({r.family for r in preds}) + decided: dict[int, Prediction] = {} + gates: dict[str, dict[str, Any]] = {} + for fold in sorted({r.fold for r in preds}): + train = [r for r in chosen if r.fold != fold] + inner: list[Prediction] = [] + for k in sorted({r.fold for r in train}): + mine = [r for r in train if r.fold == k] + inner += [replace(r, p=p) for r, p in zip(mine, normalised([r for r in train if r.fold != k], mine, **floors), strict=True) + if p is not None] # fmt: skip + t_block = pooled_threshold(inner, precision=precision) + t_quiet = {f: quiet_threshold([r for r in inner if r.family == f], recall=recall) for f in families} + held = [(i, r) for i, r in enumerate(preds) if r.fold == fold] + for (i, r), p in zip(held, normalised(train, [r for _, r in held], **floors), strict=True): + quiet = t_quiet[r.family] + if p is not None and t_block is not None and r.blockable and p >= t_block: + point = "BLOCK" + elif p is None or quiet is None or p >= quiet: + point = "ADVISORY" + else: + point = "DROP" + decided[i] = replace(r, p=p, point=point) + gates[fold] = {"t_block": t_block, "t_quiet": t_quiet, "training_rows": len(train), "normalised_training_rows": len(inner)} + return [decided[i] for i in range(len(preds))], gates + + +def family_floor(rows: Sequence[Prediction], *, floor: int = FAMILY_FLOOR) -> dict[str, dict[str, Any]]: + """Per family, its held-out ``BLOCK`` flags among ``rows`` (the primary population, every fold): it may block only + when none is wrong and there are at least ``floor``; otherwise its flags are reported as ADVISORY.""" + out: dict[str, dict[str, Any]] = {} + for family in sorted({r.family for r in rows}): + flagged = [r for r in rows if r.family == family and r.point == "BLOCK"] + errors = sum(1 for r in flagged if r.label == 0) + out[family] = {"flagged": len(flagged), "errors": errors, "floor": floor, "may_block": errors == 0 and len(flagged) >= floor} + return out + + +def apply_floor(rows: Sequence[Prediction], floors: Mapping[str, Mapping[str, Any]]) -> list[Prediction]: + """``BLOCK`` becomes ``ADVISORY`` in every family that does not pass :func:`family_floor`.""" + return [replace(r, point="ADVISORY") if r.point == "BLOCK" and not floors.get(r.family, {}).get("may_block") else r for r in rows] + + def _at(rows: Sequence[Prediction], level: str) -> dict[str, float | None]: flagged = [r for r in rows if r.point in (("BLOCK",) if level == "block" else ("BLOCK", "ADVISORY"))] positives = sum(r.label for r in rows) @@ -408,7 +500,8 @@ def candidate_from_example( __all__ = [ - "BOUNDS", "PARSE_FAILURE_LIMIT", "TARGET_PRECISION", "TARGET_RECALL", "Points", "Prediction", "calibrated_block", + "BOUNDS", "FAMILY_FLOOR", "PARSE_FAILURE_LIMIT", "TARGET_PRECISION", "TARGET_RECALL", "Points", "Prediction", "apply_floor", + "calibrated_block", "family_floor", "normalised", "pooled_block_gate", "pooled_threshold", "quiet_threshold", "candidate_from_example", "choose_points", "cluster_bootstrap", "evaluate", "explain", "floor_status", "incompatible", "learning_curve", "nested_points", "nested_precision_block", "precision_threshold", "predict", "strata", "summary", "targets", "unevaluable_families", diff --git a/src/openultrasast/learn/experiments.py b/src/openultrasast/learn/experiments.py index a62ca0e..24947bd 100644 --- a/src/openultrasast/learn/experiments.py +++ b/src/openultrasast/learn/experiments.py @@ -44,7 +44,20 @@ import yaml from ..plane.budget import BudgetExhausted, MeteredClient from ..plane.memory import MemoryStore, row_id from .compile import load_program, spec_of -from .evaluate import BOUNDS, TARGET_PRECISION, Prediction, calibrated_block, candidate_from_example, nested_precision_block, wilson_lower +from .evaluate import ( + BOUNDS, + FAMILY_FLOOR, + TARGET_PRECISION, + TARGET_RECALL, + Prediction, + apply_floor, + calibrated_block, + candidate_from_example, + family_floor, + nested_precision_block, + pooled_block_gate, + wilson_lower, +) from .examples import EXAMPLE_KIND, Example from .folds import Fold, compile_split, folds_digest, outer_folds from .program import Caller, Decision, Program, ProgramSpec @@ -60,12 +73,17 @@ RUN_TASK = "run" ARM_KINDS = ("program", "plane", "rule") # decision-rule experiments (kind: rule): one replay-only scoring pass of the incumbent programs, arms differ only in # the BLOCK rule applied to the same scores -RULES = ("calibrated_block", "precision_bound_block") +RULES = ("calibrated_block", "precision_bound_block", "pooled_block_gate") +NORMALISATIONS = ("platt",) # pooled_block_gate: how the scores of different families are put on one scale (Req 6.4) +RULE_PARAMETERS: Mapping[str, set[str]] = { + "precision_bound_block": {"precision", "bound", "min_flagged", "select_on"}, + "pooled_block_gate": {"precision", "recall", "floor", "normalise", "select_on"}, +} POPULATIONS = ("paired", "pooled") # paired: units whose candidate has both sides present; pooled: every unit SCORE_ARM = "score" # the arm name of a rule experiment's outcome rows (the shared scores) MISS_VERDICT = "replay_miss" # an outcome row of a unit excluded because a response was not cached # rule metrics, per family and population of held-out (outer-fold) units: direction only (no paired test) -RULE_METRICS: Mapping[str, int] = {"flagged_precision_lb": 1, "flagged_precision": 1, "coverage": 1, "flagged": 1} +RULE_METRICS: Mapping[str, int] = {"flagged_precision_lb": 1, "flagged_precision": 1, "coverage": 1, "flagged": 1, "quiet_recall": 1} UNITS = ("candidate", "file", "repository") PAIRINGS = ("paired", "assigned") # O'Brien-Fleming boundaries for two equally spaced looks at alpha 0.05, two-sided (design section 6). @@ -277,17 +295,24 @@ def _check_rule_manifest(path: Path, raw: Mapping[str, Any], arms: Mapping[str, name = arm.rule.get("name") if name not in RULES: raise ExperimentError(f"{path}: arm {arm.name} rule.name {name!r} (one of {', '.join(RULES)})") - allowed = {"name"} | ({"precision", "bound", "min_flagged", "select_on"} if name == "precision_bound_block" else set()) + allowed = {"name"} | RULE_PARAMETERS.get(str(name), set()) if set(arm.rule) - allowed: raise ExperimentError(f"{path}: arm {arm.name} rule parameters {sorted(set(arm.rule) - allowed)} are not understood") if arm.rule.get("bound", "wilson") not in BOUNDS or arm.rule.get("select_on", "paired") not in POPULATIONS: raise ExperimentError(f"{path}: arm {arm.name} bound in {BOUNDS} and select_on in {POPULATIONS}") + if name == "pooled_block_gate": + if arm.rule.get("normalise") not in NORMALISATIONS: + raise ExperimentError(f"{path}: arm {arm.name} normalise in {NORMALISATIONS} (declared before the run)") + if int(arm.rule.get("floor", FAMILY_FLOOR)) < 1 or not 0 < float(arm.rule.get("recall", TARGET_RECALL)) <= 1: + raise ExperimentError(f"{path}: arm {arm.name} floor >= 1 and recall in (0, 1]") populations = dict(raw.get("populations") or {}) if populations.get("primary") not in POPULATIONS or any(p not in POPULATIONS for p in populations.get("secondary", [])): raise ExperimentError(f"{path}: populations.primary (and secondary) in {POPULATIONS}") decision = dict(raw.get("decision") or {}) if not 0 < float(decision.get("target_precision", 0)) <= 1 or decision.get("arm") not in arms: raise ExperimentError(f"{path}: decision.target_precision in (0, 1] and decision.arm naming an arm") + if decision.get("pooled") and arms[str(decision["arm"])].rule.get("name") != "pooled_block_gate": + raise ExperimentError(f"{path}: decision.pooled judges a pooled_block_gate arm only") # --- register ------------------------------------------------------------------------------------------------------------ @@ -1087,6 +1112,17 @@ def _apply_rule(arm: Arm, preds: Sequence[Prediction], seed: int) -> tuple[list[ if arm.rule["name"] == "calibrated_block": return calibrated_block(preds, seed=seed) select_on = str(arm.rule.get("select_on", "paired")) + if arm.rule["name"] == "pooled_block_gate": # the family floor is applied by analyse_rule, on held-out flags + rows, gates = pooled_block_gate( + preds, eligible=(lambda r: bool(r.pair)) if select_on == "paired" else None, + precision=float(arm.rule.get("precision", TARGET_PRECISION)), recall=float(arm.rule.get("recall", TARGET_RECALL)), + ) # fmt: skip + info = { + f: {"t_block_by_fold": {j: g["t_block"] for j, g in gates.items()}, + "t_quiet_by_fold": {j: g["t_quiet"].get(f) for j, g in gates.items()}} + for f in sorted({r.family for r in preds}) + } # fmt: skip + return rows, info decided, thresholds = nested_precision_block( preds, eligible=(lambda r: bool(r.pair)) if select_on == "paired" else None, precision=float(arm.rule.get("precision", TARGET_PRECISION)), bound=str(arm.rule.get("bound", "wilson")), @@ -1095,11 +1131,38 @@ def _apply_rule(arm: Arm, preds: Sequence[Prediction], seed: int) -> tuple[list[ return decided, {f: {"thresholds_by_fold": t} for f, t in thresholds.items()} -def decide_rule(manifest: Manifest, families: Mapping[str, Mapping[str, Any]]) -> tuple[str, str]: +def decide_pooled(manifest: Manifest, families: Mapping[str, Mapping[str, Any]], overall: Mapping[str, Any]) -> tuple[str, str]: + """The pre-registered rule of a pooled gate (Req 6.4, exp-004), on the primary population's held-out flags of + every family together, before the family floor: **reject** when they flag at a point precision below the target; + **adopt** when their Wilson 95% lower bound reaches the target and at least one family passes the floor (no + held-out error, at least ``floor`` flags) -- those families block, the others stay advisory; otherwise + **inconclusive** (the incumbent rule stays).""" + decision = dict(manifest.raw["decision"]) + arm, target = str(decision["arm"]), float(decision["target_precision"]) + primary = str(manifest.raw["populations"]["primary"]) + cell = overall[arm][primary] + if cell["flagged"] and not cell["meets_target_point"]: + return "reject", f"{arm} flags held-out {primary} units of all families together below precision {target}" + passing = sorted(f for f, v in families.items() if v[arm]["floor"]["may_block"]) + if cell["meets_target_lb"] and passing: + return "adopt", f"{arm}'s pooled held-out Wilson lower bound reaches {target}; {', '.join(passing)} pass the family floor and block" + if not cell["flagged"]: + return "inconclusive", f"{arm} offers no BLOCK on held-out {primary} units; the incumbent stays" + if not cell["meets_target_lb"]: + return "inconclusive", f"{arm}'s pooled held-out Wilson lower bound is below {target}; the incumbent stays" + return "inconclusive", f"{arm} reaches {target} pooled but no family passes the floor; every family stays advisory" + + +def decide_rule(manifest: Manifest, families: Mapping[str, Mapping[str, Any]], overall: Mapping[str, Any] | None = None) -> tuple[str, str]: """The pre-registered rule: **reject** the decision arm when it flags held-out units of the primary population in some family at a point precision below the target; **adopt** it when it does not and some family's held-out - Wilson lower bound reaches the target; otherwise **inconclusive** (the incumbent rule stays).""" + Wilson lower bound reaches the target; otherwise **inconclusive** (the incumbent rule stays). A manifest whose + ``decision.pooled`` is set is judged by :func:`decide_pooled` instead.""" decision = dict(manifest.raw["decision"]) + if decision.get("pooled"): + if overall is None: + raise ExperimentError(f"{manifest.id}: a pooled decision needs the pooled-over-families totals") + return decide_pooled(manifest, families, overall) arm, target = str(decision["arm"]), float(decision["target_precision"]) primary = str(manifest.raw["populations"]["primary"]) cells = {f: v[arm][primary] for f, v in families.items()} @@ -1142,6 +1205,19 @@ def analyse_rule( return [r for r in rows if r.pair] if population == "paired" else list(rows) decided = {name: _apply_rule(arm, preds, manifest.seed) for name, arm in manifest.arms.items()} + gated = {name for name, arm in manifest.arms.items() if arm.rule.get("name") == "pooled_block_gate"} + floors = { # held-out flags of the primary population decide which families may block; no threshold moves + name: {f: family_floor([r for r in select(decided[name][0], populations[0]) if r.family == f], + floor=int(manifest.arms[name].rule.get("floor", FAMILY_FLOOR))).get(f) + or {"flagged": 0, "errors": 0, "floor": int(manifest.arms[name].rule.get("floor", FAMILY_FLOOR)), "may_block": False} + for f in manifest.families} + for name in gated + } # fmt: skip + + def quiet_recall(rows: Sequence[Prediction]) -> float | None: + positives = [r for r in rows if r.label == 1] + return round(sum(1 for r in positives if r.point in ("BLOCK", "ADVISORY")) / len(positives), 4) if positives else None + families: dict[str, dict[str, Any]] = {} for family in manifest.families: cell: dict[str, Any] = {} @@ -1149,6 +1225,9 @@ def analyse_rule( mine = [r for r in rows if r.family == family] cell[name] = {**info.get(family, {}), **{p: flag_summary(select(mine, p), target, resamples=n, seed=manifest.seed) for p in populations}} # fmt: skip + if name in gated: + cell[name]["floor"] = floors[name][family] + cell[name]["quiet_recall"] = {p: quiet_recall(select(mine, p)) for p in populations} mine = [r for r in preds if r.family == family] folds = sorted({r.fold for r in mine}) cell["capacity"] = { @@ -1160,7 +1239,12 @@ def analyse_rule( name: {p: flag_summary(select(rows, p), target, resamples=n, seed=manifest.seed) for p in populations} for name, (rows, _) in decided.items() } - verdict, reason = decide_rule(manifest, families) + after_floor = { # what users would get: BLOCK only in the families that pass the floor (secondary, not deciding) + name: {p: flag_summary(select(apply_floor(decided[name][0], floors[name]), p), target, resamples=n, seed=manifest.seed) + for p in populations} + for name in sorted(gated) + } # fmt: skip + verdict, reason = decide_rule(manifest, families, overall) spend = {SCORE_ARM: round(sum(o.usd or 0.0 for o in outcomes if o.arm == SCORE_ARM), 6)} return { "experiment": manifest.id, "manifest_digest": manifest.digest, "hypothesis": manifest.hypothesis, @@ -1172,7 +1256,8 @@ def analyse_rule( "populations": {"primary": populations[0], "secondary": populations[1:]}, "target_precision": target, "min_flagged_for_bound": min_flagged_for_bound(target), "primary": [m.as_dict() for m in manifest.primary], "secondary": [m.as_dict() for m in manifest.secondary], "families": families, "overall": overall, "resamples": n, - "decision": verdict, "reason": reason, "spend_usd": spend, **dict(provenance or {}), + **({"overall_after_floor": after_floor} if after_floor else {}), "decision": verdict, "reason": reason, "spend_usd": spend, + **dict(provenance or {}), } # fmt: skip diff --git a/tests/test_learn_rules.py b/tests/test_learn_rules.py index 6eda3ee..14cfa06 100644 --- a/tests/test_learn_rules.py +++ b/tests/test_learn_rules.py @@ -4,13 +4,24 @@ today's calibrated BLOCK rule, scored once from the response cache at zero cost from __future__ import annotations import json +from dataclasses import replace from pathlib import Path import pytest from test_learn_experiments import CODE, EXAMPLES, PRICES, _json_shape, artifact, committed, example_rows, excerpt_text, paid, put_program from openultrasast.learn import experiments as ex -from openultrasast.learn.evaluate import Prediction, calibrated_block, nested_precision_block, precision_threshold, wilson_lower +from openultrasast.learn.evaluate import ( + Prediction, + apply_floor, + calibrated_block, + family_floor, + nested_precision_block, + pooled_block_gate, + precision_threshold, + quiet_threshold, + wilson_lower, +) # fmt: skip from openultrasast.learn.program import Candidate, Program from openultrasast.learn.retrieve import Target from openultrasast.plane.memory import FileStore @@ -245,3 +256,123 @@ def test_the_registered_exp003_manifest_loads() -> None: assert manifest.kind == "rule" and manifest.budget_total_usd == 0 and manifest.raw["populations"]["primary"] == "paired" assert manifest.arms["A"].rule == {"name": "calibrated_block"} and manifest.arms["B"].rule["bound"] == "wilson" assert len(manifest.families) == 6 and set(manifest.programs) == set(manifest.families) + + +# --- the pooled BLOCK gate with a family floor (Req 6.4 amended 2026-10-03; exp-004) ------------------------------------------- + +LOW_FLOORS = {"per_family_floor": 2, "pooled_floor": 1} # small synthetic families still get a Platt map + + +def two_scales(per_fold: int = 12, *, error: tuple[int, str] | None = None) -> list[Prediction]: + """Two families on different raw scales: injection positives at 0.9, negatives at 0.1; path at 0.7 and 0.3. With + ``error = (fold, family)`` one negative of that fold and family scores like a positive and says vulnerable.""" + rows = [] + for f in range(5): + for family, hi, lo in (("injection", 0.9, 0.1), ("path", 0.7, 0.3)): + for i in range(per_fold): + key = f"{family}{f}-{i}" + bad = error == (f, family) and i == 0 + rows.append(pred(f, 1, hi, family=family, group=f"org/{key}", verdict="vulnerable", pair=key)) + verdict = "vulnerable" if bad else "not_vulnerable" + rows.append(pred(f, 0, hi if bad else lo, family=family, group=f"org/{key}", verdict=verdict, pair=key)) + return rows + + +def test_one_pooled_threshold_blocks_where_no_family_alone_reaches_the_bound() -> None: + rows = two_scales() # 48 training positives per family: fewer than the 73 the bound needs, 96 together + _, alone = nested_precision_block(rows) + assert all(t is None for fam in alone.values() for t in fam.values()) + decided, gates = pooled_block_gate(rows, **LOW_FLOORS) + assert all(g["t_block"] is not None for g in gates.values()) + blocked = [r for r in decided if r.point == "BLOCK"] + assert {r.family for r in blocked} == {"injection", "path"} and all(r.label == 1 for r in blocked) + assert sum(r.label for r in blocked) == sum(r.label for r in rows) # every positive, on both raw scales + + +def test_no_held_out_label_moves_a_threshold_or_the_map() -> None: + """Fold outer-0's labels are flipped (and its scores kept): its t_block, t_quiet and normalised scores must not move, + while the other folds, which train on outer-0, do see the change.""" + rows = two_scales() + flipped = [replace(r, label=1 - r.label) if r.fold == "outer-0" else r for r in rows] + decided, gates = pooled_block_gate(rows, **LOW_FLOORS) + again, regates = pooled_block_gate(flipped, **LOW_FLOORS) + assert regates["outer-0"] == gates["outer-0"] + assert [r.p for r in decided if r.fold == "outer-0"] == [r.p for r in again if r.fold == "outer-0"] + assert [r.point for r in decided if r.fold == "outer-0"] == [r.point for r in again if r.fold == "outer-0"] + assert any(regates[f"outer-{f}"] != gates[f"outer-{f}"] for f in range(1, 5)) # the test has teeth + + +def test_a_family_with_one_held_out_error_cannot_block() -> None: + rows = two_scales(16, error=(2, "path")) + decided, _ = pooled_block_gate(rows, **LOW_FLOORS) + floors = family_floor(decided, floor=10) + assert floors["path"]["errors"] == 1 and floors["path"]["flagged"] >= 10 and floors["path"]["may_block"] is False + assert floors["injection"] == {"flagged": 80, "errors": 0, "floor": 10, "may_block": True} + shipped = apply_floor(decided, floors) + assert not any(r.point == "BLOCK" for r in shipped if r.family == "path") + assert all(r.point == "ADVISORY" for r, d in zip(shipped, decided, strict=True) if d.point == "BLOCK" and r.family == "path") + few = [r for r in decided if r.family == "injection"][:18] # 9 clean flags: below the floor + assert family_floor(few, floor=10)["injection"] == {"flagged": 9, "errors": 0, "floor": 10, "may_block": False} + + +def test_quiet_threshold_keeps_ninety_percent_of_positives_and_nothing_is_silenced_blind() -> None: + rows = [replace(pred(0, 1, 0.5), p=v / 10) for v in range(1, 11)] + assert quiet_threshold(rows) == 0.2 # 9 of 10 positives at or above 0.2; 0.3 keeps 8 + assert quiet_threshold([replace(pred(0, 0, 0.5), p=0.9)]) is None + lone = [pred(0, 1, 0.9, family="deserialization", group="org/lone")] # a family with no training rows: no map, no t_quiet + decided, _ = pooled_block_gate([*two_scales(), *lone], **LOW_FLOORS) + assert decided[-1].p is None and decided[-1].point == "ADVISORY" + assert any(r.point == "DROP" for r in decided) # low scores below t_quiet are silent + + +POOLED_MANIFEST = RULE_MANIFEST.replace( + " B: {{label: bound, rule: {{name: precision_bound_block, bound: wilson, precision: 0.95, select_on: paired}}}}\n" + " C: {{label: point, rule: {{name: precision_bound_block, bound: point, precision: 0.95, min_flagged: 2, select_on: paired}}}}", + " B: {{label: gate, rule: {{name: pooled_block_gate, precision: 0.95, recall: 0.9, floor: 10, normalise: platt, select_on: paired}}}}", +).replace("decision: {{arm: B, target_precision: 0.95}}", "decision: {{arm: B, target_precision: 0.95, pooled: true}}") + + +def pooled_outcomes(error: tuple[int, str] | None) -> tuple[list[ex.Outcome], list[ex.Unit]]: + units, outcomes = [], [] + for r in two_scales(16, error=error): + u = ex.Unit(f"{r.pair}-{r.label}", r.group, r.family, r.label, r.fold, r.pair) + units.append(u) + outcomes.append(ex.Outcome(u.unit, u.group, u.family, u.label, u.pair, ex.SCORE_ARM, 0, r.s, r.verdict, 0.0)) + return outcomes, units + + +def test_the_pooled_verdict_follows_the_registered_rule(tmp_path: Path) -> None: + path = tmp_path / "exp-904-pooled.yaml" + path.write_text(POOLED_MANIFEST.format(id="exp-904-pooled", pid="0" * 64, units=tmp_path / "u.jsonl")) + manifest = ex.load_manifest(path) + assert manifest.arms["B"].rule["normalise"] == "platt" + outcomes, units = pooled_outcomes(None) + report = ex.analyse_rule(manifest, outcomes, units, resamples=20) + assert report["decision"] == "adopt" and report["overall"]["B"]["paired"]["meets_target_lb"] + assert all(report["families"][f]["B"]["floor"]["may_block"] for f in ("injection", "path")) + assert report["families"]["path"]["B"]["quiet_recall"]["paired"] == 1.0 + outcomes, units = pooled_outcomes((2, "path")) + report = ex.analyse_rule(manifest, outcomes, units, resamples=20) + assert report["families"]["path"]["B"]["floor"]["may_block"] is False and report["decision"] == "adopt" + assert "injection" in report["reason"] and "path" not in report["reason"] + assert report["overall_after_floor"]["B"]["paired"]["flagged"] == report["families"]["injection"]["B"]["paired"]["flagged"] + text = path.read_text() + for bad, why in ( + (text.replace("normalise: platt", "normalise: rank"), "normalise"), + (text.replace("floor: 10", "floor: 0"), "floor"), + (text.replace("name: pooled_block_gate", "name: precision_bound_block"), "not understood"), + ): + path.write_text(bad) + with pytest.raises(ex.ExperimentError, match=why): + ex.load_manifest(path) + + +def test_the_exp004_manifest_registers_the_procedure_before_its_units() -> None: + path = ROOT / "plane/experiments/exp-004-pooled-block-gate.yaml" + if not path.exists(): + pytest.skip("exp-004 is not written yet") + manifest = ex.load_manifest(path) + gate = manifest.arms["B"].rule + assert manifest.kind == "rule" and manifest.budget_total_usd == 0 and manifest.raw["decision"]["pooled"] is True + assert gate["name"] == "pooled_block_gate" and gate["normalise"] == "platt" and gate["floor"] == 10 and gate["precision"] == 0.95 + assert manifest.raw["populations"]["primary"] == "paired" and set(manifest.programs) == set(manifest.families)