diff --git a/.kiro/specs/learned-decision-engine/tasks.md b/.kiro/specs/learned-decision-engine/tasks.md index 261af21..896d636 100644 --- a/.kiro/specs/learned-decision-engine/tasks.md +++ b/.kiro/specs/learned-decision-engine/tasks.md @@ -69,7 +69,7 @@ a model or embedding call. - Tests: `test_learn_program.py` (no `EXCLUDED_FIELDS` in prompts, prefix identity, parsing, `unsure` cannot BLOCK, budget ceiling, replay). - _Requirements: 3.1, 3.3_ - - [ ] 6.5 Compile, calibrate, evaluate (no model in tests) + - [x] 6.5 Compile, calibrate, evaluate (no model in tests) - `learn/compile.toml` (pre-registered metric: balanced Brier score; seeds; budget), `learn/compile.py` (bootstrap few-shot on `C_boot`, 6 proposed + baseline instructions scored on `C_val`, 4 demonstration sets, alternates for leave-one-framework-out, artifact `programs/.json`), `learn/calibrate.py` (cross-fitted diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index 7f95b28..e222f4a 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 150, + "modules": 153, "counts": { - "load_bearing": 143, + "load_bearing": 146, "standalone_capability": 7 }, "orphaned": [], @@ -114,6 +114,7 @@ "cli", "harness", "hunter_tools", + "learn.cli", "model.endpoint", "pairs", "plane.memory", @@ -352,8 +353,11 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "cli", + "learn.calibrate", "learn.cli", + "learn.compile", "learn.embeddings", + "learn.evaluate", "learn.examples", "learn.excerpt", "learn.features", @@ -366,6 +370,14 @@ "plane.tasks.roles" ] }, + { + "module": "learn.calibrate", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "learn.evaluate" + ] + }, { "module": "learn.cli", "classification": "load_bearing", @@ -374,6 +386,15 @@ "cli" ] }, + { + "module": "learn.compile", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "learn.cli", + "learn.evaluate" + ] + }, { "module": "learn.embeddings", "classification": "load_bearing", @@ -383,12 +404,23 @@ "learn.retrieve" ] }, + { + "module": "learn.evaluate", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "learn.cli", + "learn.compile" + ] + }, { "module": "learn.examples", "classification": "load_bearing", "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "learn.cli", + "learn.compile", + "learn.evaluate", "learn.folds", "learn.program", "learn.retrieve" @@ -415,8 +447,11 @@ { "module": "learn.folds", "classification": "load_bearing", - "reason": "imported by learn.program, learn.retrieve", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ + "learn.cli", + "learn.compile", + "learn.evaluate", "learn.program", "learn.retrieve" ] @@ -427,7 +462,10 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "cli", + "learn.calibrate", "learn.cli", + "learn.compile", + "learn.evaluate", "learn.examples", "learn.excerpt", "learn.retrieve" @@ -437,13 +475,18 @@ "module": "learn.program", "classification": "load_bearing", "reason": "learned-decision-engine Req 3.1/3.3: the AI classifier program (signature, Retrieve -> Classify(k), response cache); `ousast learn compile` (task 6.5) and the decide task (task 8) are its callers", - "importers": [] + "importers": [ + "learn.cli", + "learn.compile", + "learn.evaluate" + ] }, { "module": "learn.retrieve", "classification": "load_bearing", - "reason": "imported by learn.program", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ + "learn.evaluate", "learn.program" ] }, @@ -462,6 +505,8 @@ "classification": "load_bearing", "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ + "learn.compile", + "learn.evaluate", "learn.features", "learn.program", "learn.retrieve", @@ -511,6 +556,7 @@ "cli", "cpg.backend", "cpg.session", + "learn.cli", "learn.features", "learn.labels", "model.candidates", @@ -610,6 +656,7 @@ "reason": "model-grounded-detection Req 8: the judge client, wired by model/judge.py in group 3", "importers": [ "cli", + "learn.cli", "plane.budget", "plane.generate", "push.assistance", @@ -795,6 +842,7 @@ "cli", "improve.memory", "learn.cli", + "learn.compile", "learn.embeddings", "learn.examples", "learn.program", @@ -818,6 +866,7 @@ "reason": "ai-service-plane Req 2.3/2.4: metered chat client with usd and calls ceilings; task integration in tasks 3, 6", "importers": [ "learn.cli", + "learn.compile", "learn.embeddings", "learn.program", "plane.generate", @@ -876,6 +925,7 @@ "importers": [ "improve.memory", "learn.cli", + "learn.compile", "learn.embeddings", "learn.examples", "learn.program", @@ -1088,6 +1138,7 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "learn.cli", + "learn.compile", "learn.program", "model.endpoint", "tool_hunter" @@ -1099,6 +1150,7 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "learn.cli", + "learn.compile", "learn.program", "model.endpoint", "tool_hunter" diff --git a/src/openultrasast/learn/calibrate.py b/src/openultrasast/learn/calibrate.py new file mode 100644 index 0000000..e46f5cf --- /dev/null +++ b/src/openultrasast/learn/calibrate.py @@ -0,0 +1,230 @@ +"""Calibration of the program's score on held-out repositories (learned-decision-engine design 4.4, Req 3.5, 7.4). + +Pure Python, no fitted classifier: the raw score ``s`` (the vote mean) is mapped by **Platt scaling** on +``logit(clip(s, 0.01, 0.99))``, a two-parameter logistic fitted by Newton's method with Platt's smoothed targets. + +- **Cross-fitted**: the map applied to fold j is fitted on the out-of-fold predictions of the other folds only + (:func:`cross_fit`), so no map is checked on the fold it was fitted on. +- Per family when the family reaches the per-family floor of positive and negative repository groups; pooled with + a per-family intercept between the floors; otherwise ``insufficient data`` (no calibrated probability). +- **The check** (:func:`check`): a reliability table (5 quantile bins), the calibration slope and intercept (a + logistic regression of the label on ``logit(p)``) with repository-bootstrap 95% intervals, and the expected + calibration error. It holds when the slope interval contains 1, the intercept interval contains 0 and ECE <= 0.05; + otherwise only ADVISORY is offered (``block.offered = false, reason = "calibration"``). +""" + +from __future__ import annotations + +import math +import random +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any + +from .labels import PER_FAMILY_FLOOR, POOLED_FLOOR + +CLIP = 0.01 +ECE_LIMIT = 0.05 +BINS = 5 + + +def logit(p: float) -> float: + q = min(max(p, CLIP), 1 - CLIP) + return math.log(q / (1 - q)) + + +def sigmoid(z: float) -> float: + if z >= 0: + return 1 / (1 + math.exp(-z)) + e = math.exp(z) + return e / (1 + e) + + +def _solve(a: list[list[float]], b: list[float]) -> list[float]: + """Gaussian elimination with partial pivoting (the systems here are 2 to ~10 wide).""" + n = len(b) + m = [row[:] + [b[i]] for i, row in enumerate(a)] + for col in range(n): + pivot = max(range(col, n), key=lambda r: abs(m[r][col])) + m[col], m[pivot] = m[pivot], m[col] + if abs(m[col][col]) < 1e-12: + continue + for r in range(n): + if r != col: + f = m[r][col] / m[col][col] + for c in range(col, n + 1): + m[r][c] -= f * m[col][c] + return [m[i][n] / m[i][i] if abs(m[i][i]) >= 1e-12 else 0.0 for i in range(n)] + + +def fit_logistic(xs: Sequence[Sequence[float]], ys: Sequence[float], *, l2: float = 1e-3, iterations: int = 100) -> list[float]: + """Weights of ``y ~ sigmoid(w . x)`` by Newton's method with a small ridge (separable data stays finite).""" + d = len(xs[0]) if xs else 0 + w = [0.0] * d + for _ in range(iterations): + grad = [-l2 * wi for wi in w] + hess = [[(l2 if i == j else 0.0) for j in range(d)] for i in range(d)] + for x, y in zip(xs, ys, strict=True): + p = sigmoid(sum(wi * xi for wi, xi in zip(w, x, strict=True))) + for i in range(d): + grad[i] += (y - p) * x[i] + for j in range(d): + hess[i][j] += p * (1 - p) * x[i] * x[j] + step = _solve(hess, grad) + w = [wi + si for wi, si in zip(w, step, strict=True)] + if max((abs(s) for s in step), default=0.0) < 1e-9: + break + return w + + +def platt_targets(ys: Sequence[int]) -> list[float]: + pos = sum(1 for y in ys if y == 1) + neg = len(ys) - pos + hi, lo = (pos + 1) / (pos + 2), 1 / (neg + 2) + return [hi if y == 1 else lo for y in ys] + + +def fit_platt(scores: Sequence[float], labels: Sequence[int]) -> tuple[float, float]: + """(a, b) of ``p = sigmoid(a logit(s) + b)``.""" + a, b = fit_logistic([[logit(s), 1.0] for s in scores], platt_targets(labels)) + return a, b + + +@dataclass(frozen=True) +class Row: + """One out-of-fold prediction: its fold, family, repository group, label and raw score.""" + + fold: str + family: str + group: str + label: int + s: float + + +@dataclass(frozen=True) +class CalibrationMap: + per_family: Mapping[str, tuple[float, float]] = field(default_factory=dict) + pooled: tuple[float, float] | None = None # slope and baseline intercept of the pooled map + intercepts: Mapping[str, float] = field(default_factory=dict) # per-family intercept shift of the pooled map + insufficient: tuple[str, ...] = () + + def apply(self, family: str, s: float) -> float | None: + if family in self.per_family: + a, b = self.per_family[family] + return sigmoid(a * logit(s) + b) + if self.pooled is not None and family in self.intercepts: + a, b = self.pooled + return sigmoid(a * logit(s) + b + self.intercepts[family]) + return None + + def as_dict(self) -> dict[str, Any]: + return { + "per_family": {f: list(v) for f, v in sorted(self.per_family.items())}, "pooled": list(self.pooled) if self.pooled else None, + "intercepts": dict(sorted(self.intercepts.items())), "insufficient": list(self.insufficient), + } # fmt: skip + + +def _groups(rows: Sequence[Row], family: str, label: int) -> int: + return len({r.group for r in rows if r.family == family and r.label == label}) + + +def fit_map(rows: Sequence[Row], *, per_family_floor: int = PER_FAMILY_FLOOR, pooled_floor: int = POOLED_FLOOR) -> CalibrationMap: + families = sorted({r.family for r in rows}) + level = {f: min(_groups(rows, f, 1), _groups(rows, f, 0)) for f in families} + own = [f for f in families if level[f] >= per_family_floor] + pooled = [f for f in families if pooled_floor <= level[f] < per_family_floor] + per_family = {f: fit_platt([r.s for r in rows if r.family == f], [r.label for r in rows if r.family == f]) for f in own} + pooled_map: tuple[float, float] | None = None + intercepts: dict[str, float] = {} + if pooled: + chosen = [r for r in rows if r.family in pooled] + xs = [[logit(r.s), 1.0] + [1.0 if r.family == f else 0.0 for f in pooled[1:]] for r in chosen] + w = fit_logistic(xs, platt_targets([r.label for r in chosen])) + pooled_map = (w[0], w[1]) + intercepts = {f: (w[2 + i - 1] if i else 0.0) for i, f in enumerate(pooled)} + return CalibrationMap(per_family, pooled_map, intercepts, tuple(f for f in families if level[f] < pooled_floor)) + + +def cross_fit(rows: Sequence[Row], **floors: int) -> tuple[list[float | None], dict[str, CalibrationMap]]: + """Calibrated ``p`` per row, each from a map fitted on the other folds' rows only; and the map per fold.""" + maps = {fold: fit_map([r for r in rows if r.fold != fold], **floors) for fold in sorted({r.fold for r in rows})} + return [maps[r.fold].apply(r.family, r.s) for r in rows], maps + + +@dataclass(frozen=True) +class Check: + reliability: tuple[Mapping[str, float], ...] + slope: float + slope_ci: tuple[float, float] + intercept: float + intercept_ci: tuple[float, float] + ece: float + holds: bool + n: int + + def as_dict(self) -> dict[str, Any]: + return { + "reliability": [dict(b) for b in self.reliability], "slope": round(self.slope, 4), + "slope_ci": [round(v, 4) for v in self.slope_ci], "intercept": round(self.intercept, 4), + "intercept_ci": [round(v, 4) for v in self.intercept_ci], "ece": round(self.ece, 4), "holds": self.holds, "n": self.n, + } # fmt: skip + + +def _slope_intercept(ps: Sequence[float], ys: Sequence[int]) -> tuple[float, float]: + a, b = fit_logistic([[logit(p), 1.0] for p in ps], [float(y) for y in ys]) + return a, b + + +def reliability(ps: Sequence[float], ys: Sequence[int], bins: int = BINS) -> tuple[list[dict[str, float]], float]: + order = sorted(range(len(ps)), key=lambda i: ps[i]) + table, ece = [], 0.0 + for b in range(bins): + idx = order[b * len(order) // bins : (b + 1) * len(order) // bins] + if not idx: + continue + mean_p = sum(ps[i] for i in idx) / len(idx) + rate = sum(ys[i] for i in idx) / len(idx) + table.append({"n": len(idx), "mean_p": round(mean_p, 4), "positive_rate": round(rate, 4)}) + ece += len(idx) / len(ps) * abs(mean_p - rate) + return table, ece + + +def _percentile(values: Sequence[float], q: float) -> float: + ordered = sorted(values) + return ordered[min(len(ordered) - 1, max(0, round(q * (len(ordered) - 1))))] + + +def check(ps: Sequence[float], ys: Sequence[int], groups: Sequence[str], *, resamples: int = 200, seed: int = 0) -> Check: + """The calibration check on held-out predictions, with repository-cluster bootstrap intervals.""" + table, ece = reliability(ps, ys) + slope, intercept = _slope_intercept(ps, ys) + by_group: dict[str, list[int]] = {} + for i, g in enumerate(groups): + by_group.setdefault(g, []).append(i) + names = sorted(by_group) + rng = random.Random(seed) + slopes, intercepts = [], [] + for _ in range(resamples): + idx = [i for g in (rng.choice(names) for _ in names) for i in by_group[g]] + sample_y = [ys[i] for i in idx] + if len(set(sample_y)) < 2: + continue + a, b = _slope_intercept([ps[i] for i in idx], sample_y) + slopes.append(a) + intercepts.append(b) + slope_ci = (_percentile(slopes, 0.025), _percentile(slopes, 0.975)) if slopes else (math.nan, math.nan) + intercept_ci = (_percentile(intercepts, 0.025), _percentile(intercepts, 0.975)) if intercepts else (math.nan, math.nan) + holds = bool(slopes) and slope_ci[0] <= 1 <= slope_ci[1] and intercept_ci[0] <= 0 <= intercept_ci[1] and ece <= ECE_LIMIT + return Check(tuple(table), slope, slope_ci, intercept, intercept_ci, ece, holds, len(ps)) + + +def offered(result: Check) -> dict[str, dict[str, Any]]: + """What a family is offered: ADVISORY always; BLOCK only when calibration holds.""" + block = {"offered": True, "reason": None} if result.holds else {"offered": False, "reason": "calibration"} + return {"block": block, "advisory": {"offered": True, "reason": None}} + + +__all__ = [ + "ECE_LIMIT", "CalibrationMap", "Check", "Row", "check", "cross_fit", "fit_logistic", "fit_map", "fit_platt", "logit", "offered", + "reliability", "sigmoid", +] # fmt: skip diff --git a/src/openultrasast/learn/cli.py b/src/openultrasast/learn/cli.py index f792359..f388f2b 100644 --- a/src/openultrasast/learn/cli.py +++ b/src/openultrasast/learn/cli.py @@ -34,6 +34,24 @@ def add_commands(learn_sub: Any) -> None: embed.add_argument("--budget-calls", type=int) embed.add_argument("--batch", type=int, default=32) embed.add_argument("--dry-run", action="store_true", help="count and estimate the tokens; no call") + for name, text in ( + ("compile", "compile the program on the compile split (bootstrap few-shot, instruction search)"), + ("evaluate", "evaluate a compiled program on the outer folds: cross-fitted calibration, nested operating points"), + ("curve", "the program's metric over memory size on a fixed candidate subset"), + ): + sub = learn_sub.add_parser(name, help=text) + sub.add_argument("--memory", help="the store (default OUSAST_MEMORY, then results_root()/memory)") + sub.add_argument("--profile", choices=("static", "plane"), default="static") + sub.add_argument("--spec", type=Path, help="the pre-registered compile spec (default: the packaged learn/compile.toml)") + sub.add_argument("--manifest", type=Path, default=Path("plane/models/deepseek-flash.yaml"), help="the chat Model with the prices") + sub.add_argument("--budget-usd", type=float, help="the ceiling of a paid run; required unless --replay-only") + sub.add_argument("--budget-calls", type=int) + sub.add_argument("--replay-only", action="store_true", help="no client: every response must come from the cache") + sub.add_argument("--out", type=Path, help="write the report (counts and intervals only) here") + if name != "compile": + sub.add_argument("--program", required=True, help="the compiled program id (programs/.json in the store)") + if name == "curve": + sub.add_argument("--subset", type=int, default=200) def run(args: argparse.Namespace) -> int: @@ -41,10 +59,104 @@ def run(args: argparse.Namespace) -> int: return _memory_build(args) if args.learn_command == "memory" and args.memory_command == "embed": return _memory_embed(args) + if args.learn_command in ("compile", "evaluate", "curve"): + return _program_command(args) print(f"learn: unknown command {args.learn_command}", file=sys.stderr) return 2 +def _caller(args: argparse.Namespace, store: Any, model: str) -> Any: + """The metered, cached caller; ``None`` client under --replay-only. Exits 2 without a ceiling or a key.""" + from ..plane.budget import MeteredClient + from .program import Caller + + parameters = _model_parameters(args.manifest) + if args.replay_only: + return Caller(None, model, parameters, store=store) + if args.budget_usd is None: + raise SystemExit("learn: --budget-usd is required (a paid run starts only under a ceiling), or --replay-only") + from ..config import load_config + from ..model.endpoint import resolve_chat_endpoint + + resolved = resolve_chat_endpoint(load_config(None)) + if resolved is None: + raise SystemExit("learn: no chat endpoint configured (DEEPSEEK_API_KEY); nothing was spent") + client = MeteredClient(resolved[0], prices=parameters, budget_usd=args.budget_usd, budget_calls=args.budget_calls) + return Caller(client, model, parameters, store=store) + + +def _identities(store: Any) -> dict[str, tuple[str, ...]]: + """Per example id, the strings its demonstration's rationale must not name: group, repository and path parts.""" + from .compile import group_identities + from .examples import EXAMPLE_KIND + + out: dict[str, tuple[str, ...]] = {} + for record in store.rows(kind=EXAMPLE_KIND): + row = record.row + path = str(row.get("candidate", "")).partition("::")[0] + out[str(row["id"])] = (*group_identities(str(row.get("group") or "")), str(row["repo"]), path, path.rsplit("/", 1)[-1]) + return out + + +def _program_command(args: argparse.Namespace) -> int: + """compile | evaluate | curve. Exit 3 when the ceiling stops the run (``unfinished``, no artifact, no number).""" + from datetime import date + + from ..plane.budget import BudgetExhausted + from ..plane.memory import open_store + from .compile import DEFAULT_SPEC, compile_program, load_program, load_spec, spec_of + from .embeddings import EmbeddingCache + from .evaluate import candidate_from_example, evaluate, learning_curve, predict, targets, unevaluable_families + from .examples import load_examples + from .folds import compile_split, outer_folds + from .program import Program + + store = open_store(args.memory) + spec = load_spec(args.spec or DEFAULT_SPEC) + memory = load_examples(store, args.profile) + cache = EmbeddingCache(store) + vectors = {sha: v for sha in cache.names() if (v := cache.get(sha)) is not None} or None + + def excerpt_text(sha: str) -> str | None: + data = store.get_blob("excerpts", sha) + return data.decode("utf-8") if data is not None else None + + model = str(spec["classify"].get("model", "deepseek-flash")) + caller = _caller(args, store, model) + report: dict[str, Any] + try: + if args.learn_command == "compile": + ids = _identities(store) + result = compile_program( + memory, caller, excerpt_text, spec=spec, profile=args.profile, vectors=vectors, + identities=lambda e: ids.get(e.id, ()), store=store, created=date.today().isoformat(), + ) # fmt: skip + report = {"status": result.status, "program_id": result.program_id, "calls": result.calls, "replayed": result.replayed, + "reason": result.reason, "bootstrap": result.bootstrap} # fmt: skip + else: + artifact = load_program(store, args.program) + program = Program(spec_of(artifact), memory, excerpt_text, vectors) + split = compile_split(memory, fraction=float(spec["split"].get("compile_fraction", 0.25)), seed=spec.seed) + folds = outer_folds(memory, seed=spec.seed, exclude=split.groups) + pairs = targets(folds, memory, skip=unevaluable_families(memory)) + candidate_of = candidate_from_example(excerpt_text, vectors) + if args.learn_command == "evaluate": + report = evaluate(predict(program, folds, caller, candidate_of, seed=spec.seed, pairs=pairs), seed=spec.seed) + else: + report = {"curve": learning_curve(program, pairs[: args.subset], caller, candidate_of, seed=spec.seed)} + report["program_id"] = args.program + except BudgetExhausted as exc: + print(f"learn {args.learn_command}: unfinished -- {exc}", file=sys.stderr) + return 3 + report["spend"] = {"usd": caller.usd(), "client_calls": caller.calls, "replayed": caller.replayed} + text = json.dumps(report, indent=2, sort_keys=True, default=str) + "\n" + if args.out is not None: + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(text, encoding="utf-8") + print(text, end="") + return 0 + + def _model_parameters(path: Path) -> dict[str, Any]: from ..plane.manifests import load_manifests diff --git a/src/openultrasast/learn/compile.py b/src/openultrasast/learn/compile.py new file mode 100644 index 0000000..da7ae5e --- /dev/null +++ b/src/openultrasast/learn/compile.py @@ -0,0 +1,526 @@ +"""Compilation of the program, DSPy-style and in-house (learned-decision-engine design 4.3, 5, Req 3.3). + +``ousast learn compile --profile P --spec learn/compile.toml`` produces a compiled program from a frozen memory +snapshot. Nothing is fitted; the instruction and the demonstrations are *chosen* against the pre-registered metric +of ``compile.toml`` (the balanced Brier score of the raw score on ``C_val``): + +1. **Compile split** (:func:`.folds.compile_split`): 25% of the groups, never evaluated, split into ``C_boot`` and + ``C_val``; during the compile the memory is the compile split only, so no evaluated repository shapes it. +2. **Bootstrap few-shot**: the program without demonstrations (retrieval on, ``k = 1``, temperature 0) on up to 150 + candidates of ``C_boot``; the kept demonstrations are right, confident (>= 0.7), cite a line in range and name no + identity -- no repository, no path, no CVE/GHSA id (:func:`identity_hits`). The rationale was produced without + the label, so a demonstration teaches reasoning, not the answer. Four seeded sets of three (a positive, a + negative, one of either), each demonstration with an alternate of another framework for leave-one-framework-out. +3. **Instruction search**: one proposal call (the signature, a signal glossary, 20 summarised bootstrap errors -- no + code) for 6 instructions; with the baseline, 7 are scored with demo set 1, then the best with sets 2-4. +4. The **artifact** is one canonical JSON (sorted keys), stored as the keyed blob ``programs/.json``, + where ``program_id`` is the sha256 of the artifact without its own id. It records the spec's sha256, the + memory snapshot digest, the folds digest and the response-cache digest; spend is priced from the cached usage, + so a replay writes the identical artifact. Whether a run replayed is reported beside it, never inside it. + +A compile that reaches its ceiling (``BudgetExhausted``) is ``unfinished`` and writes no artifact. + +**Packaging** (maintainer decision 2026-10-01): only demonstrations from permissively licensed code (MIT, BSD, +Apache-2.0, ISC, zlib) may ship in the package; :func:`packageable` swaps the others for a permissive alternate or +drops them, and the result is a new program (its own id) to be re-evaluated as its own variant. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import tomllib +from collections.abc import Callable, Iterable, Mapping, Sequence +from dataclasses import dataclass, field, replace +from pathlib import Path +from typing import Any + +from ..plane.budget import BudgetExhausted +from ..plane.memory import MemoryStore +from .examples import Example +from .folds import CompileSplit, Fold, assert_disjoint, compile_split, folds_digest, outer_folds +from .labels import _ADVISORY +from .program import ( + BASELINE_INSTRUCTION, + SIGNATURE, + Caller, + Candidate, + Decision, + Demo, + Program, + ProgramSpec, + label_word, +) +from .schema import SCHEMA_VERSION, feature_set_digest + +DEFAULT_SPEC = Path(__file__).resolve().parent / "compile.toml" +KIND = "llm-program" +PERMISSIVE = ("mit", "bsd", "0bsd", "apache-2.0", "apache 2.0", "apache2", "apache-2", "isc", "zlib") +_PATH = re.compile(r"(? Mapping[str, Any]: + return dict(self.data.get(section) or {}) + + @property + def seed(self) -> int: + return int(self.data.get("seed", 0)) + + +def load_spec(path: Path = DEFAULT_SPEC) -> CompileSpec: + text = Path(path).read_bytes() + return CompileSpec(tomllib.loads(text.decode("utf-8")), hashlib.sha256(text).hexdigest()) + + +# --- the metric ---------------------------------------------------------------------------------------------------------- + + +def balanced_brier(rows: Sequence[tuple[str, int, str, float]], *, repository_cap: int = 5) -> float | None: + """Mean over (family, label) cells of the cell's Brier score, each repository capped at ``repository_cap`` + effective rows per cell. ``rows`` are ``(family, label, group, s)``.""" + cells: dict[tuple[str, int], list[tuple[str, float]]] = {} + for family, label, group, s in rows: + cells.setdefault((family, label), []).append((group, (s - label) ** 2)) + scores = [] + for members in cells.values(): + per_group: dict[str, int] = {} + for group, _ in members: + per_group[group] = per_group.get(group, 0) + 1 + weights = [min(1.0, repository_cap / per_group[g]) for g, _ in members] + scores.append(sum(w * e for w, (_, e) in zip(weights, members, strict=True)) / sum(weights)) + return sum(scores) / len(scores) if scores else None + + +def secondary(rows: Sequence[tuple[int, float, str]], precision: float = 0.90) -> dict[str, float | None]: + """Recall at the lowest threshold reaching ``precision`` and the share of ``unsure`` (``rows``: label, s, verdict).""" + positives = sum(label for label, _, _ in rows) + recall = None + for t in sorted({s for _, s, _ in rows}): + chosen = [label for label, s, _ in rows if s >= t] + if chosen and sum(chosen) / len(chosen) >= precision: + recall = sum(chosen) / positives if positives else None + break + share = sum(1 for _, _, v in rows if v == "unsure") / len(rows) if rows else None + return {"recall_at_precision_0.90": recall, "unsure_share": share} + + +# --- identities ---------------------------------------------------------------------------------------------------------- + + +def identity_hits(rationale: str, identities: Iterable[str] = ()) -> list[str]: + """Identities a rationale names: CVE/GHSA ids, paths and source file names, and any given repository string.""" + hits = [m.group(0) for m in _ADVISORY.finditer(rationale)] + [m.group(0) for m in _PATH.finditer(rationale)] + folded = rationale.casefold() + hits += [i for i in identities if len(i) >= 4 and i.casefold() in folded] + return hits + + +def group_identities(group: str) -> tuple[str, ...]: + parts = [p for p in re.split(r"[/:]", group) if p] + return (group, *parts) + + +def permissive(license_id: str) -> bool: + text = license_id.strip().casefold().removesuffix(" license") + return bool(text) and any(text == p or text.startswith(p + "-") or text.startswith(p + " ") for p in PERMISSIVE) + + +# --- bootstrap, demo sets, proposals ---------------------------------------------------------------------------------------- + + +def _order(items: Iterable[Example], seed: int, salt: str) -> list[Example]: + return sorted(items, key=lambda e: hashlib.sha256(f"{seed}\x00{salt}\x00{e.id}".encode()).hexdigest()) + + +@dataclass +class Bootstrap: + pool: list[Demo] = field(default_factory=list) + errors: list[dict[str, Any]] = field(default_factory=list) + rejected: dict[str, int] = field(default_factory=dict) + + def reject(self, why: str) -> None: + self.rejected[why] = self.rejected.get(why, 0) + 1 + + +def _signal_summary(example: Example) -> dict[str, Any]: + ran = {k: v for k, v in example.x.items() if v not in (None, 0, False) and not k.startswith("lang.")} + return dict(sorted(ran.items())[:12]) + + +def bootstrap( + program: Program, + examples: Sequence[Example], + fold: Fold, + caller: Caller, + candidate_of: Callable[[Example], Candidate | None], + *, + min_confidence: float, + identities: Callable[[Example], Iterable[str]], + seed: int = 0, +) -> Bootstrap: + out = Bootstrap() + for example in examples: + candidate = candidate_of(example) + if candidate is None: + out.reject("no excerpt") + continue + decision = program.decide(candidate, fold, caller, seed=seed) + answer = decision.answers[0] + right = decision.verdict == label_word(example.label) + if not right: + out.errors.append({"family": example.family, "label": label_word(example.label), "verdict": decision.verdict, + "confidence": answer.confidence, "signals": _signal_summary(example)}) # fmt: skip + out.reject("wrong or unsure") + continue + if answer.parse_failed or answer.confidence < min_confidence: + out.reject("not confident") + continue + if identity_hits(answer.rationale, identities(example)): + out.reject("rationale names an identity") + continue + out.pool.append( + Demo( + example.id, + example.excerpt_sha, + example.label, + example.family, + answer.verdict, + answer.confidence, + answer.rationale, + answer.cited_lines, + ) # fmt: skip + ) + return out + + +def draw_demo_sets( + pool: Sequence[Demo], examples: Mapping[str, Example], *, sets: int = 4, size: int = 3, seed: int = 0 +) -> list[tuple[Demo, ...]]: + """``sets`` seeded sets of ``size``: one positive, one negative, then any, preferring a family not yet in the set; + each demonstration carries an alternate of the same label from no shared framework (or ``None``).""" + + def ordered(salt: str, items: Iterable[Demo]) -> list[Demo]: + return sorted(items, key=lambda d: hashlib.sha256(f"{seed}\x00{salt}\x00{d.example_id}".encode()).hexdigest()) + + out: list[tuple[Demo, ...]] = [] + for n in range(sets): + positives = ordered(f"pos{n}", [d for d in pool if d.label == 1]) + negatives = ordered(f"neg{n}", [d for d in pool if d.label == 0]) + chosen: list[Demo] = [*positives[:1], *negatives[:1]] + families = {d.family for d in chosen} + rest = [d for d in ordered(f"any{n}", pool) if d not in chosen] + rest.sort(key=lambda d: d.family in families) # stable: other families first + chosen += rest[: max(0, size - len(chosen))] + with_alternates = [] + for demo in chosen: + frameworks = set(examples[demo.example_id].frameworks) + alternate = next( + (a for a in ordered(f"alt{n}:{demo.example_id}", pool) + if a.label == demo.label and a not in chosen and not frameworks & set(examples[a.example_id].frameworks)), + None, + ) # fmt: skip + with_alternates.append(replace(demo, alternate=alternate)) + if with_alternates: + out.append(tuple(with_alternates)) + return out + + +PROPOSAL_INSTRUCTION = ( + "Propose instructions for a classifier. It reads one function's numbered code, the signals static instruments\n" + "emitted for it and similar labelled examples, and answers a JSON verdict (vulnerable, not_vulnerable, unsure)\n" + 'with a confidence and a rationale citing lines. Answer one JSON object {"instructions": []}.' +) + + +def glossary() -> str: + from .schema import FEATURES + + return "\n".join(f"{s.name}: {s.type}{' (prior)' if s.prior else ''} from {s.instrument}" for s in FEATURES) + + +def propose(caller: Caller, errors: Sequence[Mapping[str, Any]], *, n: int, shown: int) -> list[str]: + user = ( + f"Signature fields: {json.dumps([f.__dict__ for f in SIGNATURE.fields])}\nSignal glossary:\n{glossary()}\n" + f"Mistakes of the baseline instruction (signals and verdicts only):\n{json.dumps(list(errors)[:shown], sort_keys=True)}\n" + f"Baseline instruction:\n{BASELINE_INSTRUCTION}\nPropose {n} different instructions that would avoid these mistakes." + ) + messages: list[dict[str, object]] = [{"role": "system", "content": PROPOSAL_INSTRUCTION}, {"role": "user", "content": user}] + from ..provider.openrouter import OpenRouterError, parse_json_content + + try: + data = parse_json_content(caller.ask(messages, temperature=0.7, sample="propose")) + except OpenRouterError: + return [] + found = data.get("instructions") if isinstance(data, dict) else None + texts = [t.strip() for t in found if isinstance(t, str) and t.strip()] if isinstance(found, list) else [] + return list(dict.fromkeys(texts))[:n] + + +# --- scoring and the artifact ------------------------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Scored: + instruction: int # 0 = baseline, n = proposal n + demo_set: int + score: float | None + secondary: Mapping[str, float | None] + prompt_tokens: int + + def as_dict(self) -> dict[str, Any]: + return { + "instruction": self.instruction, + "demo_set": self.demo_set, + "score": None if self.score is None else round(self.score, 6), + "secondary": {k: None if v is None else round(v, 6) for k, v in self.secondary.items()}, + "prompt_tokens": self.prompt_tokens, + } + + +def score( + program: Program, examples: Sequence[Example], fold: Fold, caller: Caller, candidate_of: Callable[[Example], Candidate | None], *, + repository_cap: int, seed: int, instruction: int, demo_set: int, +) -> Scored: # fmt: skip + rows: list[tuple[str, int, str, float]] = [] + second: list[tuple[int, float, str]] = [] + tokens = 0 + for example in examples: + candidate = candidate_of(example) + if candidate is None: + continue + prompt, _ = program.prepare(candidate, fold, seed=seed) + tokens = max(tokens, round(len(prompt.prefix) / 4)) + decision: Decision = program.decide(candidate, fold, caller, seed=seed) + rows.append((example.family, example.label, example.group, decision.s)) + second.append((example.label, decision.s, decision.verdict)) + return Scored(instruction, demo_set, balanced_brier(rows, repository_cap=repository_cap), secondary(second), tokens) + + +def best(configurations: Sequence[Scored], tolerance: float) -> Scored: + valid = [c for c in configurations if c.score is not None] + if not valid: + raise CompileError("no configuration produced a score") + top = min(c.score for c in valid if c.score is not None) + close = [c for c in valid if c.score is not None and c.score <= top + tolerance] + return min(close, key=lambda c: (c.prompt_tokens, configurations.index(c))) + + +def canonical(artifact: Mapping[str, Any]) -> bytes: + return json.dumps(artifact, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + +def program_id(artifact: Mapping[str, Any]) -> str: + return hashlib.sha256(canonical({k: v for k, v in artifact.items() if k != "program_id"})).hexdigest() + + +def memory_snapshot_digest(examples: Iterable[Example]) -> str: + return hashlib.sha256("\n".join(sorted(f"{e.id}:{e.label}:{e.excerpt_sha}" for e in examples)).encode()).hexdigest() + + +@dataclass +class CompileResult: + status: str # done | unfinished + artifact: dict[str, Any] | None + program_id: str | None + calls: int # client calls made by this run (0 on a full replay) + replayed: int + reason: str | None = None + bootstrap: dict[str, Any] = field(default_factory=dict) + + +def compile_program( + memory: Sequence[Example], + caller: Caller, + excerpt_text: Callable[[str], str | None], + *, + spec: CompileSpec, + profile: str = "static", + vectors: Mapping[str, Sequence[float]] | None = None, + identities: Callable[[Example], Iterable[str]] | None = None, + families: Sequence[str] = (), + store: MemoryStore | None = None, + created: str = "", + code_commit: str = "", +) -> CompileResult: + """Compile one profile's program from ``memory`` (module docstring). Deterministic given the memory, the spec and + the response cache.""" + seed = spec.seed + pool = [e for e in memory if e.profile == profile] + split: CompileSplit = compile_split(pool, fraction=float(spec["split"].get("compile_fraction", 0.25)), + boot=float(spec["split"].get("boot_fraction", 0.6)), seed=seed) # fmt: skip + evaluation = outer_folds(pool, seed=seed, exclude=split.groups) + assert_disjoint(split, evaluation) + compile_memory = [e for e in pool if e.group in split.groups] + fold = Fold("compile", "compile", frozenset(), phase="compile") + boot = _order([e for e in compile_memory if e.group in split.boot], seed, "boot")[: int(spec["bootstrap"].get("max_candidates", 150))] + val = _order([e for e in compile_memory if e.group in split.val], seed, "val")[: int(spec["search"].get("val_candidates", 60))] + if not boot or not val: + raise CompileError(f"compile split too small: {len(boot)} bootstrap and {len(val)} validation candidates") + retrieval, classify = spec["retrieval"], spec["classify"] + base = ProgramSpec( + profile=profile, model=str(classify.get("model", "deepseek-flash")), k=int(classify.get("k", 1)), k_ret=int(retrieval.get("k", 6)), + lam=float(retrieval.get("lam", 0.5)), n1=int(retrieval.get("n1", 40)), priors=str(retrieval.get("priors", "off")), + max_output_tokens=int(classify.get("max_output_tokens", 300)), + ) # fmt: skip + from .evaluate import candidate_from_example + + candidate_of = candidate_from_example(excerpt_text, vectors) + ids = identities or (lambda e: group_identities(e.group)) + tolerance = float(spec["metric"].get("tie_tolerance", 0.005)) + cap = int(spec["metric"].get("repository_cap", 5)) + try: + boot_run = bootstrap( + Program(base, compile_memory, excerpt_text, vectors, families), boot, fold, caller, candidate_of, + min_confidence=float(spec["bootstrap"].get("min_confidence", 0.7)), identities=ids, seed=seed, + ) # fmt: skip + index = {e.id: e for e in compile_memory} + sets = draw_demo_sets(boot_run.pool, index, sets=int(spec["bootstrap"].get("demo_sets", 4)), + size=int(spec["bootstrap"].get("demos_per_set", 3)), seed=seed) # fmt: skip + if not sets: + raise CompileError(f"bootstrap kept no demonstration ({boot_run.rejected})") + shown = int(spec["search"].get("errors_shown", 20)) + instructions = [BASELINE_INSTRUCTION, *propose(caller, boot_run.errors, n=int(spec["search"].get("proposals", 6)), shown=shown)] + configurations: list[Scored] = [] + + def run(i: int, d: int) -> Scored: + program = Program(replace(base, instruction=instructions[i], demos=sets[d]), compile_memory, + excerpt_text, vectors, families) # fmt: skip + result = score(program, val, fold, caller, candidate_of, repository_cap=cap, seed=seed, instruction=i, demo_set=d) + configurations.append(result) + return result + + for i in range(len(instructions)): + run(i, 0) + chosen_instruction = best(configurations, tolerance).instruction + for d in range(1, len(sets)): + run(chosen_instruction, d) + winner = best(configurations, tolerance) + except BudgetExhausted as exc: + return CompileResult("unfinished", None, None, caller.calls, caller.replayed, reason=str(exc)) + artifact: dict[str, Any] = { + "kind": KIND, + "schema_version": SCHEMA_VERSION, + "feature_set_digest": feature_set_digest(), + "profile": profile, + "signature": {"digest": SIGNATURE.digest, "fields": [f.__dict__ for f in SIGNATURE.fields]}, + "instruction": instructions[winner.instruction], + "instruction_source": "baseline" if winner.instruction == 0 else f"proposed:{winner.instruction}", + "demos": [d.as_dict() for d in sets[winner.demo_set]], + "retrieval": { + "distance": "gower-v1", + "n1": base.n1, + "k": base.k_ret, + "lam": base.lam, + "priors": base.priors, + "near_duplicate_cos": float(retrieval.get("near_duplicate_cos", 0.98)), + "embedding_model": "openai/text-embedding-3-small" if vectors else None, + "balance": {"per_label": base.k_ret // 2, "per_group": 2, "other_families": 2}, + "excerpt": {"candidate_lines": 80, "candidate_chars": 4000, "example_lines": 40, "example_chars": 2000}, + }, + "classify": { + "model": base.model, + "model_params_digest": caller.params_digest, + "k": base.k, + "temperature": [0.0, 0.7], + "confidence": "vote_mean", + "max_output_tokens": base.max_output_tokens, + }, + "calibration": None, + "operating_points": None, + "priors": base.priors, + "roles": "deterministic", + "data": { + "memory_snapshot_digest": memory_snapshot_digest(pool), + "compile_groups_digest": hashlib.sha256("\n".join(sorted(split.groups)).encode()).hexdigest(), + "folds_digest": folds_digest(split, evaluation), + "sources": sorted({e.source for e in pool}), + "examples": len(pool), + "groups": len({e.group for e in pool}), + }, + "compile": { + "spec_sha256": spec.sha256, + "metric": str(spec["metric"].get("name", "balanced_brier_v1")), + "configurations": [c.as_dict() for c in configurations], + "usd": caller.usd(), + "requests": len(set(caller.keys)), + "usage": dict(caller.usage), + "response_cache_digest": caller.cache_digest(), + "bootstrap": {"kept": len(boot_run.pool), "rejected": dict(sorted(boot_run.rejected.items())), "candidates": len(boot)}, + }, + "seed": seed, + "code_commit": code_commit, + "created": created, + } + artifact["program_id"] = program_id(artifact) + if store is not None: + store.put_blob("programs", canonical(artifact), name=artifact["program_id"]) + return CompileResult( + "done", artifact, artifact["program_id"], caller.calls, caller.replayed, bootstrap=artifact["compile"]["bootstrap"] + ) + + +def load_program(store: MemoryStore, pid: str) -> dict[str, Any]: + data = store.get_blob("programs", pid, verify=False) + if data is None: + raise CompileError(f"no program {pid[:12]} in {store.describe()}") + artifact = dict(json.loads(data)) + if program_id(artifact) != pid: + raise CompileError(f"program {pid[:12]} does not hash to its id: the artifact changed") + return artifact + + +def spec_of(artifact: Mapping[str, Any]) -> ProgramSpec: + retrieval, classify = artifact["retrieval"], artifact["classify"] + return ProgramSpec( + instruction=str(artifact["instruction"]), + demos=tuple(Demo.from_dict(d) for d in artifact["demos"]), + profile=str(artifact["profile"]), + model=str(classify["model"]), + k=int(classify["k"]), + k_ret=int(retrieval["k"]), + lam=float(retrieval["lam"]), + n1=int(retrieval["n1"]), + priors=str(retrieval.get("priors", "off")), + max_output_tokens=int(classify.get("max_output_tokens", 300)), + ) + + +def packageable(artifact: Mapping[str, Any], examples: Mapping[str, Example]) -> tuple[dict[str, Any], dict[str, int]]: + """The artifact with only permissively licensed demonstrations: a demonstration from other code is replaced by + its alternate when that one is permissive, dropped otherwise. A new program id; re-evaluate before adoption.""" + kept: list[dict[str, Any]] = [] + tally = {"kept": 0, "swapped": 0, "dropped": 0} + for demo in artifact["demos"]: + example = examples.get(str(demo["example_id"])) + alternate = demo.get("alternate") + alt_example = examples.get(str(alternate["example_id"])) if isinstance(alternate, Mapping) else None + if example is not None and permissive(example.license): + kept.append({**demo, "alternate": None}) + tally["kept"] += 1 + elif isinstance(alternate, Mapping) and alt_example is not None and permissive(alt_example.license): + kept.append({**alternate, "alternate": None}) + tally["swapped"] += 1 + else: + tally["dropped"] += 1 + packaged = {k: v for k, v in artifact.items() if k != "program_id"} + packaged["demos"] = kept + packaged["packaged"] = {"from": artifact.get("program_id"), "licences": list(PERMISSIVE), **tally} + packaged["program_id"] = program_id(packaged) + return packaged, tally + + +__all__ = [ + "DEFAULT_SPEC", "PERMISSIVE", "Bootstrap", "CompileError", "CompileResult", "CompileSpec", "Scored", "balanced_brier", "best", + "bootstrap", "canonical", "compile_program", "draw_demo_sets", "group_identities", "identity_hits", "load_program", "load_spec", + "memory_snapshot_digest", "packageable", "permissive", "program_id", "propose", "score", "secondary", "spec_of", +] # fmt: skip diff --git a/src/openultrasast/learn/compile.toml b/src/openultrasast/learn/compile.toml new file mode 100644 index 0000000..e346412 --- /dev/null +++ b/src/openultrasast/learn/compile.toml @@ -0,0 +1,44 @@ +# The pre-registered compile specification of the decision engine's program (learned-decision-engine design 4.3). +# Committed before any compile runs: the metric, the seeds, the search space and the budget are fixed here, and the +# compiled artifact records this file's sha256. Changing anything below after looking at a result is a new spec. +version = 1 +seed = 0 + +[metric] +name = "balanced_brier_v1" # Brier score of the raw score s on C_val, each (family, label) cell weighted equally +repository_cap = 5 # each repository counts at most 5 effective rows per cell +unsure_score = 0.5 # unsure and parse_failed answers score as s = 0.5 +tie_tolerance = 0.005 # within this, fewer prompt tokens wins, then the earlier configuration +secondary = ["recall_at_precision_0.90", "unsure_share"] # reported, never optimised + +[split] +compile_fraction = 0.25 # of the repository groups, per primary family; never an evaluation fold +boot_fraction = 0.6 # C_boot (demonstrations) vs C_val (scoring) +retrieval = "compile_only" # during the compile, memory is the compile split only: no evaluated group shapes it + +[bootstrap] +max_candidates = 150 +min_confidence = 0.7 +demo_sets = 4 +demos_per_set = 3 # one positive, one negative, one of either; families mixed + +[search] +proposals = 6 # instructions proposed in one call, scored with the hand-written baseline (7) +errors_shown = 20 # summarised bootstrap errors in the proposal prompt (signals and verdicts, no code) +val_candidates = 60 + +[classify] +model = "deepseek-flash" +k = 1 # k and lam are A/B variants (design section 6), not searched here +max_output_tokens = 300 + +[retrieval] +n1 = 40 +k = 6 +lam = 0.5 +near_duplicate_cos = 0.98 +priors = "off" + +[budget] +usd = 3.0 # a compile that reaches it is `unfinished` and writes no artifact +calls = 1200 diff --git a/src/openultrasast/learn/evaluate.py b/src/openultrasast/learn/evaluate.py new file mode 100644 index 0000000..e15e7dc --- /dev/null +++ b/src/openultrasast/learn/evaluate.py @@ -0,0 +1,333 @@ +"""Evaluation of a compiled program on repositories it never saw (learned-decision-engine design 4.4-4.5, Req 5.2, +7.3, 7.4). + +For each evaluation fold the program decides every candidate of the fold's groups with memory and demonstrations +filtered by the fold's boundary. The raw scores are then **cross-fitted** (:mod:`.calibrate`), and the operating +points are chosen on the *other* folds' calibrated predictions (nested), never on the fold reported: + +- **BLOCK**: the lowest threshold whose precision has a Wilson 95% lower bound >= 0.95 (delta units when the data + has them), counting only candidates whose majority verdict is not ``unsure``; none -> ``precision_unreachable``; +- **ADVISORY**: the highest threshold whose recall is >= 0.90; its precision is reported, not assumed; +- **DROP** below ADVISORY. + +Metrics carry repository-cluster bootstrap 95% intervals. A family or language below the pooled floor of positive +and negative repository groups is reported as ``insufficient data``, never as a number. More than 5% parse failures +make the run an instrument failure, not a result. +""" + +from __future__ import annotations + +import hashlib +import math +import random +import re +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 .examples import Example +from .folds import Fold +from .labels import PER_FAMILY_FLOOR, POOLED_FLOOR +from .program import Caller, Candidate, Program +from .retrieve import Target +from .schema import BY_NAME, SCHEMA_VERSION, feature_set_digest + +TARGET_PRECISION = 0.95 +TARGET_RECALL = 0.90 +PARSE_FAILURE_LIMIT = 0.05 +BOOTSTRAP = 2000 +_SIGNAL = re.compile(r"\b[a-z]+(?:\.[a-z_]+)+\b") + + +@dataclass(frozen=True) +class Prediction: + fold: str + family: str + group: str + label: int + s: float + verdict: str + unit: str = "pin" + language: str = "other" + source: str = "" + parse_failed: int = 0 + p: float | None = None + point: str | None = None # BLOCK | ADVISORY | DROP, from points chosen on the other folds + + @property + def blockable(self) -> bool: + return self.verdict != "unsure" + + +def wilson_lower(successes: int, n: int, z: float = 1.959964) -> float: + if n == 0: + return 0.0 + phat = successes / n + centre = phat + z * z / (2 * n) + margin = z * math.sqrt(phat * (1 - phat) / n + z * z / (4 * n * n)) + return (centre - margin) / (1 + z * z / n) + + +@dataclass(frozen=True) +class Points: + block: float | None + block_offered: bool + block_reason: str | None + advisory: float | None + + def as_dict(self) -> dict[str, Any]: + return { + "block": { + "threshold": self.block, + "offered": self.block_offered, + "reason": self.block_reason, + "target_precision": TARGET_PRECISION, + }, + "advisory": {"threshold": self.advisory, "target_recall": TARGET_RECALL}, + } + + +def choose_points(preds: Sequence[Prediction], *, precision: float = TARGET_PRECISION, recall: float = TARGET_RECALL) -> Points: + """Operating points on calibrated predictions (``p`` set); rows without ``p`` are not decided.""" + rows = [r for r in preds if r.p is not None] + thresholds = sorted({float(r.p) for r in rows if r.p is not None}) + deltas = [r for r in rows if r.unit == "delta"] + blockable = [r for r in (deltas or rows) if r.blockable] + block = None + for t in thresholds: + chosen = [r for r in blockable if (r.p or 0.0) >= t] + if chosen and wilson_lower(sum(r.label for r in chosen), len(chosen)) >= precision: + block = t + break + positives = sum(r.label for r in rows) + advisory = None + for t in reversed(thresholds): + if positives and sum(r.label for r in rows if (r.p or 0.0) >= t) / positives >= recall: + advisory = t + break + return Points(block, block is not None, None if block is not None else "precision_unreachable", advisory) + + +def nested_points(preds: Sequence[Prediction]) -> tuple[list[Prediction], dict[str, Points]]: + """Each fold's rows decided with the points chosen on the other folds' calibrated predictions.""" + points = {fold: choose_points([r for r in preds if r.fold != fold]) for fold in sorted({r.fold for r in preds})} + out = [] + for r in preds: + pts = points[r.fold] + if r.p is None: + out.append(r) + continue + if pts.block is not None and r.blockable and r.p >= pts.block: + point = "BLOCK" + elif pts.advisory is not None and r.p >= pts.advisory: + point = "ADVISORY" + else: + point = "DROP" + out.append(replace(r, point=point)) + return out, points + + +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) + benign = [r for r in rows if r.label == 0 and r.unit == "delta"] + return { + "recall": sum(r.label for r in flagged) / positives if positives else None, + "precision": sum(r.label for r in flagged) / len(flagged) if flagged else None, + "false_per_100_benign_deltas": 100 * sum(1 for r in benign if r.point == "BLOCK") / len(benign) + if benign and level == "block" + else None, + } + + +def cluster_bootstrap( + rows: Sequence[Prediction], stat: Callable[[Sequence[Prediction]], float | None], *, resamples: int = BOOTSTRAP, seed: int = 0 +) -> tuple[float, float] | None: + by_group: dict[str, list[Prediction]] = {} + for r in rows: + by_group.setdefault(r.group, []).append(r) + names = sorted(by_group) + if not names: + return None + rng = random.Random(seed) + values = [] + for _ in range(resamples): + sample = [r for g in (rng.choice(names) for _ in names) for r in by_group[g]] + value = stat(sample) + if value is not None: + values.append(value) + if not values: + return None + values.sort() + return values[int(0.025 * (len(values) - 1))], values[int(0.975 * (len(values) - 1))] + + +def summary(rows: Sequence[Prediction], *, resamples: int = BOOTSTRAP, seed: int = 0) -> dict[str, Any]: + out: dict[str, Any] = {"candidates": len(rows), "groups": len({r.group for r in rows})} + + def statistic(level: str, name: str) -> Callable[[Sequence[Prediction]], float | None]: + return lambda sample: _at(sample, level)[name] + + for level in ("block", "advisory"): + point = _at(rows, level) + out[level] = { + name: {"value": value, "ci": cluster_bootstrap(rows, statistic(level, name), resamples=resamples, seed=seed)} + for name, value in point.items() + if value is not None + } + return out + + +def floor_status(rows: Sequence[Prediction]) -> str: + pos = len({r.group for r in rows if r.label == 1}) + neg = len({r.group for r in rows if r.label == 0}) + level = min(pos, neg) + return "per_family" if level >= PER_FAMILY_FLOOR else "pooled" if level >= POOLED_FLOOR else "insufficient data" + + +def strata(rows: Sequence[Prediction], key: Callable[[Prediction], str], **kw: Any) -> dict[str, Any]: + out: dict[str, Any] = {} + for name in sorted({key(r) for r in rows}): + part = [r for r in rows if key(r) == name] + status = floor_status(part) + out[name] = {"status": status} if status == "insufficient data" else {"status": status, **summary(part, **kw)} + return out + + +def explain(rationale: str, x: Mapping[str, Any]) -> list[str]: + """The signals the rationale names that the record holds (a name it invents is dropped).""" + named = dict.fromkeys(m.group(0) for m in _SIGNAL.finditer(rationale)) + return [n for n in named if n in BY_NAME and n in x and x[n] is not None] + + +def incompatible(artifact: Mapping[str, Any], record: Mapping[str, Any], signature_digest: str) -> str | None: + """Why a program may not decide a record (the caller falls back to today's behaviour and says so).""" + if artifact.get("schema_version") != SCHEMA_VERSION or record.get("schema_version") != SCHEMA_VERSION: + return "schema version differs" + if artifact.get("feature_set_digest") != feature_set_digest() or record.get("feature_set_digest") != feature_set_digest(): + return "feature set differs" + if (artifact.get("signature") or {}).get("digest") != signature_digest: + return "signature differs" + if artifact.get("profile") != record.get("profile"): + return "profile differs" + return None + + +def targets(folds: Sequence[Fold], examples: Sequence[Example], *, skip: Iterable[str] = ()) -> list[tuple[Example, Fold]]: + """(example, fold) for every example of every usable fold's evaluated groups; ``skip`` families are never asked + (the families below the floor: no money is spent on them).""" + skipped = set(skip) + return [(e, f) for f in folds if f.status == "ok" for e in examples if e.group in f.eval_groups and e.family not in skipped] + + +def predict( + program: Program, + folds: Sequence[Fold], + caller: Caller, + candidate_of: Callable[[Any], Candidate | None], + *, + seed: int = 0, + pairs: Sequence[tuple[Example, Fold]] | None = None, +) -> list[Prediction]: + """Decide every example of every fold's evaluated groups (or the given ``pairs``), each under its own fold.""" + out: list[Prediction] = [] + for example, fold in pairs if pairs is not None else targets(folds, program.memory): + candidate = candidate_of(example) + if candidate is None: + continue + decision = program.decide(candidate, fold, caller, seed=seed) + out.append( + Prediction( + fold.name, example.family, example.group, example.label, decision.s, decision.verdict, example.unit, example.language, + example.source, decision.parse_failed, + ) + ) # fmt: skip + return out + + +def unevaluable_families(examples: Sequence[Example], *, floor: int = POOLED_FLOOR) -> list[str]: + """Families below the pooled floor of positive and negative repository groups: reported, never evaluated.""" + out = [] + for family in sorted({e.family for e in examples}): + pos = len({e.group for e in examples if e.family == family and e.label == 1}) + neg = len({e.group for e in examples if e.family == family and e.label == 0}) + if min(pos, neg) < floor: + out.append(family) + return out + + +def learning_curve( + program: Program, + pairs: Sequence[tuple[Example, Fold]], + caller: Caller, + candidate_of: Callable[[Any], Candidate | None], + *, + fractions: Sequence[float] = (0.0, 0.25, 0.5, 1.0), + seed: int = 0, +) -> list[dict[str, Any]]: + """The metric as a function of memory size: a seeded share of the memory's groups, the same fixed candidates.""" + from .compile import balanced_brier + + groups = sorted({e.group for e in program.memory}, key=lambda g: hashlib.sha256(f"{seed}\x00curve\x00{g}".encode()).hexdigest()) + points = [] + for fraction in fractions: + kept = set(groups[: int(len(groups) * fraction + 0.5)]) + sub = replace(program, memory=[e for e in program.memory if e.group in kept]) + preds = predict(sub, [], caller, candidate_of, seed=seed, pairs=pairs) + brier = balanced_brier([(p.family, p.label, p.group, p.s) for p in preds]) + points.append({"memory_fraction": fraction, "memory_groups": len(kept), "candidates": len(preds), "balanced_brier": brier}) + return points + + +def evaluate(preds: Sequence[Prediction], *, resamples: int = BOOTSTRAP, seed: int = 0, **floors: int) -> dict[str, Any]: + """Cross-fitted calibration, its check, nested operating points and the report (counts and intervals only).""" + failed = sum(r.parse_failed for r in preds) + answers = len(preds) or 1 + rows = [Row(r.fold, r.family, r.group, r.label, r.s) for r in preds] + ps, maps = cross_fit(rows, **floors) + calibrated = [replace(r, p=p) for r, p in zip(preds, ps, strict=True)] + decided = [r for r in calibrated if r.p is not None] + checked = ( + check([float(r.p) for r in decided if r.p is not None], [r.label for r in decided], [r.group for r in decided], seed=seed) + if decided + else None + ) + nested, points = nested_points(calibrated) + report: dict[str, Any] = { + "status": "instrument failure" if failed / answers > PARSE_FAILURE_LIMIT else "ok", + "parse_failed_share": round(failed / answers, 4), + "calibration": checked.as_dict() if checked else None, + "offered": offered(checked) if checked else {"block": {"offered": False, "reason": "insufficient data"}}, + "points_by_fold": {f: p.as_dict() for f, p in points.items()}, + "maps_by_fold": {f: m.as_dict() for f, m in maps.items()}, + "overall": summary([r for r in nested if r.p is not None], resamples=resamples, seed=seed), + "by_family": strata(nested, lambda r: r.family, resamples=resamples, seed=seed), + "by_language": strata(nested, lambda r: r.language, resamples=resamples, seed=seed), + "undecided": sum(1 for r in nested if r.p is None), + } + if checked is not None and not checked.holds: + report["overall"]["block"] = {"offered": False, "reason": "calibration"} + return report + + +def candidate_from_example( + excerpt_text: Callable[[str], str | None], vectors: Mapping[str, Sequence[float]] | None = None +) -> Callable[[Any], Candidate | None]: + """An evaluation candidate from a memory example: its record and stored excerpt (a 40-line example excerpt).""" + + def make(example: Any) -> Candidate | None: + code = excerpt_text(example.excerpt_sha) + if not code: + return None + return Candidate(Target.of(example, (vectors or {}).get(example.excerpt_sha)), code, example.language, tuple(example.roles)) + + return make + + +__all__ = [ + "PARSE_FAILURE_LIMIT", "TARGET_PRECISION", "TARGET_RECALL", "Points", "Prediction", "candidate_from_example", "choose_points", + "cluster_bootstrap", "evaluate", "explain", "floor_status", "incompatible", "learning_curve", "nested_points", "predict", "strata", + "summary", "targets", "unevaluable_families", + "wilson_lower", +] # fmt: skip diff --git a/tests/test_learn_calibrate.py b/tests/test_learn_calibrate.py new file mode 100644 index 0000000..9d9febb --- /dev/null +++ b/tests/test_learn_calibrate.py @@ -0,0 +1,62 @@ +"""Calibration (learned-decision-engine design 4.4, task 6.5): Platt on a known miscalibration, cross-fitting (fold +j's map never saw fold j), the slope/intercept/ECE check, and the ADVISORY-only downgrade.""" + +from __future__ import annotations + +import random + +import pytest + +from openultrasast.learn.calibrate import Row, check, cross_fit, fit_map, fit_platt, logit, offered, sigmoid + + +def _sample(n: int, a: float, b: float, seed: int = 0) -> tuple[list[float], list[int]]: + """Scores whose true probability is ``sigmoid(a logit(s) + b)``: a known miscalibration of ``s``.""" + rng = random.Random(seed) + scores = [rng.uniform(0.02, 0.98) for _ in range(n)] + labels = [1 if rng.random() < sigmoid(a * logit(s) + b) else 0 for s in scores] + return scores, labels + + +def test_platt_recovers_a_known_miscalibration() -> None: + scores, labels = _sample(6000, 2.0, -0.5) + a, b = fit_platt(scores, labels) + assert a == pytest.approx(2.0, abs=0.15) and b == pytest.approx(-0.5, abs=0.15) + + +def _rows(n_groups: int, a: float, b: float, *, family: str = "injection", folds: int = 5, seed: int = 0) -> list[Row]: + scores, labels = _sample(n_groups * 10, a, b, seed) + return [Row(f"f{(i // 10) % folds}", family, f"g{i // 10}", y, s) for i, (s, y) in enumerate(zip(scores, labels, strict=True))] + + +def test_cross_fitting_never_fits_a_fold_on_itself() -> None: + rows = _rows(100, 2.0, 0.0) + ps, maps = cross_fit(rows, per_family_floor=10, pooled_floor=5) + assert all(p is not None for p in ps) + for fold, mapping in maps.items(): + expected = fit_map([r for r in rows if r.fold != fold], per_family_floor=10, pooled_floor=5) + assert mapping == expected + # moving fold f0's labels changes every map except f0's own + flipped = [Row(r.fold, r.family, r.group, 1 - r.label if r.fold == "f0" else r.label, r.s) for r in rows] + _, flipped_maps = cross_fit(flipped, per_family_floor=10, pooled_floor=5) + assert flipped_maps["f0"] == maps["f0"] and flipped_maps["f1"] != maps["f1"] + + +def test_floors_per_family_pooled_and_insufficient() -> None: + rows = _rows(60, 1.0, 0.0, family="injection") + _rows(12, 1.0, 0.3, family="path", seed=1) + _rows(3, 1.0, 0, family="memory", seed=2) + mapping = fit_map(rows, per_family_floor=20, pooled_floor=5) + assert set(mapping.per_family) == {"injection"} and set(mapping.intercepts) == {"path"} and mapping.insufficient == ("memory",) + assert mapping.apply("memory", 0.7) is None and mapping.apply("path", 0.7) is not None + + +def test_the_check_holds_on_calibrated_scores_and_fails_otherwise() -> None: + scores, labels = _sample(1500, 1.0, 0.0, seed=3) + groups = [f"g{i // 10}" for i in range(len(scores))] + good = check(scores, labels, groups, resamples=40) + assert good.holds and good.ece <= 0.05 and good.slope_ci[0] <= 1 <= good.slope_ci[1] + assert offered(good) == {"block": {"offered": True, "reason": None}, "advisory": {"offered": True, "reason": None}} + overconfident = [min(0.99, max(0.01, 0.5 + 2.5 * (s - 0.5))) for s in scores] + bad = check(overconfident, labels, groups, resamples=40) + assert not bad.holds and bad.slope < 1 + assert offered(bad)["block"] == {"offered": False, "reason": "calibration"} and offered(bad)["advisory"]["offered"] + assert len(good.reliability) == 5 and sum(b["n"] for b in good.reliability) == 1500 diff --git a/tests/test_learn_compile.py b/tests/test_learn_compile.py new file mode 100644 index 0000000..72a63fc --- /dev/null +++ b/tests/test_learn_compile.py @@ -0,0 +1,193 @@ +"""Compilation (learned-decision-engine design 4.3 and 5, task 6.5): bootstrap few-shot keeps only right, confident, +identity-free demonstrations; instruction search on the pre-registered metric; the compile split stays out of every +evaluation fold; a ceiling ends the compile unfinished; a rerun from the response cache reproduces the artifact +byte for byte at $0 (``test_reproducible_from_cache``); the permissive-licence filter for packaged programs.""" + +from __future__ import annotations + +import json +from collections.abc import Iterator, Mapping, Sequence +from pathlib import Path + +import pytest +from learn_fixtures import ScriptedChat, by_id, corpus, default_answer, example +from test_plane_memory import FakeObjects, _real_minio + +from openultrasast.learn import compile as compile_module +from openultrasast.learn.compile import ( + DEFAULT_SPEC, + CompileError, + Scored, + balanced_brier, + best, + canonical, + compile_program, + draw_demo_sets, + identity_hits, + load_program, + load_spec, + packageable, + permissive, + program_id, +) +from openultrasast.learn.folds import Fold, FoldError +from openultrasast.learn.program import Caller, Demo +from openultrasast.plane.budget import MeteredClient +from openultrasast.plane.memory import FileStore, MemoryStore, MinioStore + +PRICES = {"cache_hit_per_m": 0.014, "input_per_m": 0.44, "output_per_m": 1.32} +MEMORY = corpus(groups=24, per_group=4) +CODE = {e.excerpt_sha: f" 10 def f(q):\n 11 return {'execute(q)' if e.label else 'escape(q)'}\n" for e in MEMORY} +SPEC = load_spec(DEFAULT_SPEC) +PROPOSALS = [f"Instruction {i}: be careful with the signals." if i == 3 else f"Instruction {i}." for i in range(1, 7)] + + +def excerpt_text(sha: str) -> str | None: + return CODE.get(sha) + + +def answer(messages: Sequence[Mapping[str, object]], temperature: float | None) -> str: + """Proposals on request; a better calibrated confidence under the 'careful' instruction, so the search moves.""" + system = str(messages[0]["content"]) + if system.startswith("Propose instructions"): + return json.dumps({"instructions": PROPOSALS}) + data = json.loads(default_answer(messages, temperature)) + if "careful" in system: + data["confidence"] = 0.95 + return json.dumps(data) + + +@pytest.fixture(params=["file", "fake-minio", "minio"]) +def store(request: pytest.FixtureRequest, tmp_path: Path) -> Iterator[MemoryStore]: + if request.param == "file": + yield FileStore(tmp_path / "memory") + elif request.param == "minio": + yield from _real_minio() + else: + yield MinioStore(FakeObjects("works"), "bucket", "p") + + +def _caller(chat: ScriptedChat, store: MemoryStore | None = None, **budget: object) -> Caller: + return Caller(MeteredClient(chat, prices=PRICES, **budget), "deepseek-flash", PRICES, store=store) # type: ignore[arg-type] + + +def _compile(caller: Caller, store: MemoryStore | None = None, memory: Sequence = MEMORY) -> compile_module.CompileResult: + return compile_program(memory, caller, excerpt_text, spec=SPEC, store=store, created="2026-10-01", code_commit="test") + + +def test_reproducible_from_cache(store: MemoryStore) -> None: + first_chat = ScriptedChat(answer) + first = _compile(_caller(first_chat, store), store) + assert first.status == "done" and first.calls > 0 and first.calls == len(first_chat.calls) + second_chat = ScriptedChat(answer) + second = _compile(_caller(second_chat, store), store) + assert second.calls == 0 and second_chat.calls == [] and second.replayed > 0 + assert first.artifact is not None and second.artifact is not None + assert canonical(first.artifact) == canonical(second.artifact) and first.program_id == second.program_id + assert load_program(store, str(first.program_id)) == first.artifact + assert first.artifact["compile"]["usd"] == second.artifact["compile"]["usd"] > 0 + + +def test_the_search_picks_a_better_instruction_and_records_every_configuration() -> None: + result = _compile(_caller(ScriptedChat(answer))) + artifact = result.artifact + assert artifact is not None and artifact["instruction_source"] == "proposed:3" and "careful" in artifact["instruction"] + configurations = artifact["compile"]["configurations"] + assert len(configurations) == 7 + 3 # 7 instructions with demo set 1, then the best with sets 2-4 + baseline = configurations[0]["score"] + assert min(c["score"] for c in configurations) < baseline + assert artifact["compile"]["spec_sha256"] == SPEC.sha256 and artifact["program_id"] == program_id(artifact) + assert artifact["data"]["folds_digest"] and len(artifact["demos"]) == 3 + + +def test_bootstrap_keeps_only_right_confident_identity_free_rationales() -> None: + def naming(messages: Sequence[Mapping[str, object]], temperature: float | None) -> str: + if str(messages[0]["content"]).startswith("Propose"): + return json.dumps({"instructions": []}) + data = json.loads(default_answer(messages, temperature)) + tail = str(messages[-1]["content"]).rsplit("Candidate:", 1)[1] + if "execute" in tail: + data["rationale"] = "As in CVE-2021-1234 and app/views.py, the value reaches execute." + return json.dumps(data) + + def timid(messages: Sequence[Mapping[str, object]], temperature: float | None) -> str: + return json.dumps({**json.loads(default_answer(messages, temperature)), "confidence": 0.5}) + + with pytest.raises(CompileError, match="kept no demonstration"): + _compile(_caller(ScriptedChat(timid))) + result = _compile(_caller(ScriptedChat(naming))) + assert result.artifact is not None + rejected = result.artifact["compile"]["bootstrap"]["rejected"] + assert rejected.get("rationale names an identity", 0) > 0 + for demo in result.artifact["demos"]: + assert identity_hits(demo["rationale"]) == [] and demo["label"] == 0 # every positive named an identity + + +def test_identity_scan() -> None: + assert identity_hits("See CVE-2020-0001 here.") == ["CVE-2020-0001"] + assert identity_hits("Line 4 of views.py passes q.") == ["views.py"] + assert identity_hits("the acme helper", ("acme/webapp", "acme")) == ["acme"] + assert identity_hits("Line 11 passes the parameter to the query.", ("acme/webapp", "acme")) == [] + + +def test_the_compile_split_never_meets_an_evaluation_fold(monkeypatch: pytest.MonkeyPatch) -> None: + real = compile_module.outer_folds + + def leaking(examples, **kw): # type: ignore[no-untyped-def] + folds = real(examples, **kw) + return [*folds, Fold("outer", "leak", frozenset({e.group for e in examples}))] + + monkeypatch.setattr(compile_module, "outer_folds", leaking) + with pytest.raises(FoldError): + _compile(_caller(ScriptedChat(answer))) + + +def test_a_ceiling_ends_the_compile_unfinished_with_no_artifact(tmp_path: Path) -> None: + store = FileStore(tmp_path / "memory") + result = _compile(_caller(ScriptedChat(answer), store, budget_calls=5), store) + assert result.status == "unfinished" and result.artifact is None and result.program_id is None + assert "calls budget exhausted" in str(result.reason) and store.blob_names("programs") == [] + + +def test_balanced_brier_weights_cells_and_caps_repositories() -> None: + rows = [("a", 1, "g1", 1.0)] * 10 + [("a", 1, "g2", 0.0)] + [("a", 0, "g3", 0.0)] + # cell (a, 1): g1 has 10 rows of error 0 (capped to weight 0.5 each = 5), g2 one row of error 1 -> 1/6 + assert balanced_brier(rows, repository_cap=5) == pytest.approx((1 / 6 + 0.0) / 2) + assert balanced_brier([]) is None + + +def test_ties_go_to_fewer_prompt_tokens() -> None: + a = Scored(0, 0, 0.100, {}, 900) + b = Scored(1, 0, 0.103, {}, 500) + c = Scored(2, 0, 0.090, {}, 2000) + assert best([a, b], 0.005) == b and best([a, b, c], 0.005) == c + + +def test_demo_sets_mix_labels_and_carry_framework_alternates() -> None: + examples = [example(f"d{i}", group=f"g{i}/r", label=i % 2, family=("injection", "path")[i % 3 == 0], + frameworks=("flask",) if i < 4 else ()) for i in range(12)] # fmt: skip + pool = [Demo(e.id, e.excerpt_sha, e.label, e.family, "x", 0.9, "r", (1,)) for e in examples] + sets = draw_demo_sets(pool, by_id(examples), sets=4, size=3, seed=1) + assert len(sets) == 4 and draw_demo_sets(pool, by_id(examples), sets=4, size=3, seed=1) == sets + index = by_id(examples) + for chosen in sets: + assert {d.label for d in chosen[:2]} == {0, 1} and len(chosen) == 3 + for demo in chosen: + if demo.alternate is not None: + assert demo.alternate.label == demo.label + assert not set(index[demo.example_id].frameworks) & set(index[demo.alternate.example_id].frameworks) + + +def test_only_permissively_licensed_demonstrations_are_packaged() -> None: + assert all(permissive(x) for x in ("MIT", "BSD-3-Clause", "Apache-2.0", "ISC", "Zlib", "mit license")) + assert not any(permissive(x) for x in ("GPL-3.0", "AGPL-3.0", "LGPL-2.1", "unlicensed", "", "proprietary", "MPL-2.0")) + mit, gpl, unknown, apache = (example(n, group=f"{n}/r", label=1, license=lic) for n, lic in + (("mit", "MIT"), ("gpl", "GPL-3.0"), ("unk", ""), ("apache", "Apache-2.0"))) # fmt: skip + demo = lambda e, alt=None: {"example_id": e.id, "excerpt_sha": e.excerpt_sha, "label": 1, "rationale": "r", "alternate": alt} # noqa: E731 + artifact = {"program_id": "x", "demos": [demo(mit), demo(gpl, demo(apache)), demo(unknown, demo(gpl))], "instruction": "i"} + artifact["program_id"] = program_id(artifact) + packaged, tally = packageable(artifact, by_id([mit, gpl, unknown, apache])) + assert tally == {"kept": 1, "swapped": 1, "dropped": 1} + assert [d["example_id"] for d in packaged["demos"]] == [mit.id, apache.id] + assert packaged["program_id"] == program_id(packaged) != artifact["program_id"] + assert packaged["packaged"]["from"] == artifact["program_id"] diff --git a/tests/test_learn_decide.py b/tests/test_learn_decide.py new file mode 100644 index 0000000..b988ea9 --- /dev/null +++ b/tests/test_learn_decide.py @@ -0,0 +1,136 @@ +"""Operating points and the evaluation report (learned-decision-engine design 4.4-4.5, task 6.5): points read from +calibrated, cross-fitted predictions of the other folds; ``unsure`` never BLOCKs; explanation fields; version +mismatch falls back; ``insufficient data``; parse failures are an instrument failure. Every client is scripted.""" + +from __future__ import annotations + +import json +import random + +import pytest +from learn_fixtures import ScriptedChat, corpus + +from openultrasast.learn.evaluate import ( + Prediction, + candidate_from_example, + choose_points, + evaluate, + explain, + incompatible, + learning_curve, + nested_points, + predict, + strata, + targets, + unevaluable_families, + wilson_lower, +) +from openultrasast.learn.folds import compile_split, outer_folds +from openultrasast.learn.program import SIGNATURE, Caller, Program, ProgramSpec +from openultrasast.learn.schema import SCHEMA_VERSION, feature_set_digest +from openultrasast.plane.budget import MeteredClient + + +def _preds(n: int = 300, *, folds: int = 3, seed: int = 0, unsure_every: int = 0) -> list[Prediction]: + rng = random.Random(seed) + out = [] + for i in range(n): + label = int(rng.random() < 0.4) + p = min(0.99, max(0.01, rng.gauss(0.8 if label else 0.25, 0.12))) + verdict = "unsure" if unsure_every and i % unsure_every == 0 else ("vulnerable" if p > 0.5 else "not_vulnerable") + out.append(Prediction(f"f{i % folds}", "injection", f"g{i // 3}", label, p, verdict, "delta", "python", p=p)) + return out + + +def test_wilson_lower_bound() -> None: + assert wilson_lower(0, 0) == 0.0 and wilson_lower(10, 10) == pytest.approx(0.722, abs=0.001) + assert wilson_lower(95, 100) == pytest.approx(0.888, abs=0.001) + + +def test_points_come_from_the_other_folds_only() -> None: + preds = _preds() + decided, points = nested_points(preds) + for fold, pts in points.items(): + assert pts == choose_points([r for r in preds if r.fold != fold]) + perturbed = [Prediction(**{**r.__dict__, "label": 1 - r.label}) if r.fold == "f0" else r for r in preds] + _, again = nested_points(perturbed) + assert again["f0"] == points["f0"] and again["f1"] != points["f1"] + assert {r.point for r in decided} <= {"BLOCK", "ADVISORY", "DROP"} + + +def test_block_needs_the_wilson_bound_and_advisory_the_recall() -> None: + pts = choose_points(_preds(600)) + assert pts.advisory is not None + rows = _preds(600) + positives = sum(r.label for r in rows) + assert sum(r.label for r in rows if (r.p or 0) >= pts.advisory) / positives >= 0.90 + if pts.block is not None: + chosen = [r for r in rows if (r.p or 0) >= pts.block and r.blockable] + assert wilson_lower(sum(r.label for r in chosen), len(chosen)) >= 0.95 + noisy = [Prediction("f0", "injection", f"g{i}", i % 2, 0.6, "vulnerable", p=0.6) for i in range(50)] + unreachable = choose_points(noisy) + assert unreachable.block is None and not unreachable.block_offered and unreachable.block_reason == "precision_unreachable" + + +def test_a_majority_unsure_never_blocks() -> None: + preds = [Prediction(f"f{i % 2}", "injection", f"g{i}", 1, 0.99, "unsure", "delta", p=0.99) for i in range(40)] + preds += [Prediction(f"f{i % 2}", "injection", f"n{i}", 0, 0.01, "not_vulnerable", "delta", p=0.01) for i in range(40)] + decided, _ = nested_points(preds) + assert not any(r.point == "BLOCK" for r in decided) + assert all(r.point == "ADVISORY" for r in decided if r.label == 1) + + +def test_explanations_keep_only_signals_the_record_holds() -> None: + x = {"qr.enabled_hits": 2, "eng.findings": 0, "eng.rung_max": None} + assert explain("qr.enabled_hits and eng.findings agree; eng.rung_max and made.up say otherwise", x) == [ + "qr.enabled_hits", + "eng.findings", + ] + + +def test_a_version_mismatch_falls_back_with_the_reason() -> None: + artifact = {"schema_version": SCHEMA_VERSION, "feature_set_digest": feature_set_digest(), "signature": {"digest": SIGNATURE.digest}, + "profile": "static"} # fmt: skip + record = {"schema_version": SCHEMA_VERSION, "feature_set_digest": feature_set_digest(), "profile": "static"} + assert incompatible(artifact, record, SIGNATURE.digest) is None + assert incompatible({**artifact, "feature_set_digest": "old"}, record, SIGNATURE.digest) == "feature set differs" + assert incompatible(artifact, {**record, "schema_version": 0}, SIGNATURE.digest) == "schema version differs" + assert incompatible({**artifact, "signature": {"digest": "x"}}, record, SIGNATURE.digest) == "signature differs" + assert incompatible(artifact, {**record, "profile": "plane"}, SIGNATURE.digest) == "profile differs" + + +def test_a_family_below_the_floor_is_insufficient_data_never_a_number() -> None: + preds = _preds(90) + few = [Prediction("f0", "memory", "openssl/openssl", i % 2, 0.5, "unsure", p=0.5) for i in range(20)] + report = strata([*preds, *few], lambda r: r.family, resamples=50) + assert report["memory"] == {"status": "insufficient data"} + assert report["injection"]["status"] in ("per_family", "pooled") and "advisory" in report["injection"] + + +def _code(sha: str) -> str: + return f" 10 def f(q):\n 11 return {'execute(q)' if int(sha, 16) % 2 else 'escape(q)'}\n" + + +def test_the_evaluation_runs_end_to_end_on_scripted_answers() -> None: + memory = corpus(groups=30, per_group=4) + codes = {e.excerpt_sha: f" 10 def f(q):\n 11 return {'execute(q)' if e.label else 'escape(q)'}\n" for e in memory} + split = compile_split(memory) + folds = outer_folds(memory, exclude=split.groups) + pairs = targets(folds, memory, skip=unevaluable_families(memory, floor=3)) + assert pairs and all(e.group in f.eval_groups for e, f in pairs) + program = Program(ProgramSpec(), memory, codes.get) + chat = ScriptedChat() + caller = Caller(MeteredClient(chat), "deepseek-flash") + preds = predict(program, folds, caller, candidate_from_example(codes.get), pairs=pairs) + assert len(preds) == len(pairs) == caller.calls + caller.replayed and caller.calls == len(chat.calls) # identical prompts replay + report = evaluate(preds, resamples=50, per_family_floor=10, pooled_floor=3) + assert report["status"] == "ok" and report["parse_failed_share"] == 0 and report["calibration"] is not None + assert set(report["points_by_fold"]) == {f.name for f in folds if f.eval_groups} + curve = learning_curve(program, pairs[:20], caller, candidate_from_example(codes.get), fractions=(0.0, 1.0)) + assert [p["memory_fraction"] for p in curve] == [0.0, 1.0] and curve[0]["memory_groups"] == 0 + + +def test_more_than_five_percent_parse_failures_is_an_instrument_failure() -> None: + preds = [Prediction(f"f{i % 3}", "injection", f"g{i}", i % 2, 0.5, "unsure", parse_failed=int(i % 10 == 0)) for i in range(60)] + assert evaluate(preds, resamples=20, per_family_floor=5, pooled_floor=2)["status"] == "instrument failure" + assert json.dumps(evaluate(preds[1:10], resamples=5, per_family_floor=5, pooled_floor=2), default=str)