From f5180a14ec224654aa321b49ffe245f7bb131169 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 1 Oct 2026 10:05:19 +0200 Subject: [PATCH] feat(learn): retrieval and the evaluation boundary (decision-engine 6.3) learn/retrieve.py: stage 1 is a per-instrument Gower distance over the profile's allow-listed features with explicit missing (both ran: mean feature distance; differing states 1; same non-ran state 0; null in one 1; priors dropped with priors=off), n1 = 40 with seeded tie-breaks; stage 2 re-ranks by r = lam(1 - D) + (1 - lam)cos, signals_only without the target's vector; the balance rule takes k_ret = 6 (<= k/2 per label, <= 2 per group, <= 2 other families; `short` recorded). eligible() is the only way into a prompt: not the target's group, not an evaluated group, not the held-out source or framework, and outside deployment not a near-duplicate (cos >= 0.98). assert_boundary repeats the check on rendered ids; demonstrations of a held-out framework swap to their alternates; deployment_fold excludes a scanned repository that is a corpus group; seed_memory never hands a case its own group. learn/folds.py: compile split (25% of groups per primary family, 60/40 boot/val), outer grouped 5-fold stratified by family and label mix, leave-one-source-out, leave-one-framework-out (insufficient data below 10 groups), assert_disjoint and a folds digest; all seeded. Tests: test_learn_retrieve.py (hand-computed Gower, lam 0/0.5, balance, near-duplicates, test_boundary over outer/LOSO/LOFO folds with demonstrations), test_learn_folds.py. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- .kiro/specs/learned-decision-engine/tasks.md | 2 +- .../measurements/2026-09-08-module-audit.json | 31 +- src/openultrasast/learn/folds.py | 186 ++++++++++++ src/openultrasast/learn/retrieve.py | 277 ++++++++++++++++++ src/openultrasast/model/audit.py | 4 + tests/learn_fixtures.py | 64 ++++ tests/test_learn_folds.py | 76 +++++ tests/test_learn_retrieve.py | 175 +++++++++++ 8 files changed, 809 insertions(+), 6 deletions(-) create mode 100644 src/openultrasast/learn/folds.py create mode 100644 src/openultrasast/learn/retrieve.py create mode 100644 tests/learn_fixtures.py create mode 100644 tests/test_learn_folds.py create mode 100644 tests/test_learn_retrieve.py diff --git a/.kiro/specs/learned-decision-engine/tasks.md b/.kiro/specs/learned-decision-engine/tasks.md index 92db26c..3f93042 100644 --- a/.kiro/specs/learned-decision-engine/tasks.md +++ b/.kiro/specs/learned-decision-engine/tasks.md @@ -52,7 +52,7 @@ a model or embedding call. `ousast learn memory embed`; scripted-client tests. - **(budget go-ahead)** embedding the store: ~5.0 M tokens, ~$0.10, ceiling $1 (OpenRouter account). - _Requirements: 3.2_ - - [ ] 6.3 Retrieval and the evaluation boundary (no model) + - [x] 6.3 Retrieval and the evaluation boundary (no model) - `learn/retrieve.py`: per-instrument Gower distance with explicit missing, `n1 = 40`, embedding re-rank `r = lam(1 - D) + (1 - lam)cos`, `signals_only` mode, balance rule (`k_ret = 6`), `eligible(example, target, fold)`, near-duplicate drop in evaluation, deployment self-exclusion. diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index c79105a..694175a 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 147, + "modules": 149, "counts": { - "load_bearing": 140, + "load_bearing": 142, "standalone_capability": 7 }, "orphaned": [], @@ -357,6 +357,8 @@ "learn.examples", "learn.excerpt", "learn.features", + "learn.folds", + "learn.retrieve", "learn.roles", "plane.tasks.features", "plane.tasks.remember", @@ -376,7 +378,8 @@ "classification": "load_bearing", "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ - "learn.cli" + "learn.cli", + "learn.retrieve" ] }, { @@ -384,7 +387,9 @@ "classification": "load_bearing", "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ - "learn.cli" + "learn.cli", + "learn.folds", + "learn.retrieve" ] }, { @@ -405,6 +410,14 @@ "plane.tasks.features" ] }, + { + "module": "learn.folds", + "classification": "load_bearing", + "reason": "imported by learn.retrieve", + "importers": [ + "learn.retrieve" + ] + }, { "module": "learn.labels", "classification": "load_bearing", @@ -413,9 +426,16 @@ "cli", "learn.cli", "learn.examples", - "learn.excerpt" + "learn.excerpt", + "learn.retrieve" ] }, + { + "module": "learn.retrieve", + "classification": "load_bearing", + "reason": "learned-decision-engine Req 3.2/3.4: retrieval from the local memory and the evaluation boundary; the program (task 6.4) and `ousast learn compile` (6.5) are its callers", + "importers": [] + }, { "module": "learn.roles", "classification": "load_bearing", @@ -432,6 +452,7 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "learn.features", + "learn.retrieve", "learn.roles", "plane.tasks.remember", "plane.tasks.roles" diff --git a/src/openultrasast/learn/folds.py b/src/openultrasast/learn/folds.py new file mode 100644 index 0000000..c278c33 --- /dev/null +++ b/src/openultrasast/learn/folds.py @@ -0,0 +1,186 @@ +"""Folds by repository group (learned-decision-engine design 4.3-4.4, Req 3.4, 7.1, 8.3). + +Every split is by ``group`` (the label builder's repository group, after its alias/advisory/commit merges), across +all corpora, seeded and deterministic: + +- the **compile split** ``C``: 25% of the groups, stratified by each group's primary family, never an evaluation + fold; inside it ``C_boot`` (60%, demonstrations) and ``C_val`` (40%, instruction scoring); +- **outer** grouped K = 5 over the groups outside ``C``, stratified by primary family and label mix; +- **leave-one-source-out**: each source held out whole (its groups evaluated, the source absent from retrieval); +- **leave-one-framework-out** on F where F has >= 10 groups; below that the fold is ``insufficient data``. + +A :class:`Fold` is what :func:`.retrieve.eligible` checks every example against. :func:`assert_disjoint` is run +before a compile: the compile split and the evaluation folds share no group. +""" + +from __future__ import annotations + +import hashlib +import json +from collections import Counter +from collections.abc import Iterable, Sequence +from dataclasses import dataclass, field + +from .examples import Example + +COMPILE_FRACTION = 0.25 +BOOT_FRACTION = 0.6 +OUTER_K = 5 +FRAMEWORK_MIN_GROUPS = 10 +INSUFFICIENT = "insufficient data" + + +class FoldError(RuntimeError): + """A group in both the compile split and an evaluation fold, or in two evaluation folds of one kind.""" + + +@dataclass(frozen=True) +class Fold: + """One evaluation (or compile, or deployment) boundary: the groups evaluated, and what retrieval must not use.""" + + kind: str # outer | leave_one_source_out | leave_one_framework_out | compile | deployment + name: str + eval_groups: frozenset[str] + held_source: str | None = None + held_framework: str | None = None + phase: str = "evaluation" # evaluation | compile | deployment: the near-duplicate drop applies outside deployment + status: str = "ok" + + +@dataclass(frozen=True) +class GroupInfo: + group: str + families: Counter[str] = field(default_factory=Counter) + labels: frozenset[int] = frozenset() + sources: frozenset[str] = frozenset() + frameworks: frozenset[str] = frozenset() + + @property + def primary_family(self) -> str: + return sorted(self.families.items(), key=lambda kv: (-kv[1], kv[0]))[0][0] if self.families else "unknown" + + +@dataclass(frozen=True) +class CompileSplit: + groups: frozenset[str] + boot: frozenset[str] + val: frozenset[str] + + def fold(self) -> Fold: + """The compile phase's boundary: nothing is excluded but the target's own group.""" + return Fold("compile", "compile", frozenset(), phase="compile") + + +def group_table(examples: Iterable[Example]) -> dict[str, GroupInfo]: + families: dict[str, Counter[str]] = {} + labels: dict[str, set[int]] = {} + sources: dict[str, set[str]] = {} + frameworks: dict[str, set[str]] = {} + for example in examples: + families.setdefault(example.group, Counter())[example.family] += 1 + labels.setdefault(example.group, set()).add(example.label) + sources.setdefault(example.group, set()).add(example.source) + frameworks.setdefault(example.group, set()).update(example.frameworks) + return {g: GroupInfo(g, families[g], frozenset(labels[g]), frozenset(sources[g]), frozenset(frameworks[g])) for g in sorted(families)} + + +def _order(items: Iterable[str], seed: int, salt: str) -> list[str]: + return sorted(items, key=lambda item: hashlib.sha256(f"{seed}\x00{salt}\x00{item}".encode()).hexdigest()) + + +def _take(groups: Sequence[str], fraction: float) -> int: + return int(len(groups) * fraction + 0.5) if len(groups) > 1 else 0 # a family's only group is never taken + + +def compile_split( + examples: Sequence[Example], *, fraction: float = COMPILE_FRACTION, boot: float = BOOT_FRACTION, seed: int = 0 +) -> CompileSplit: + """25% of the groups per primary family (seeded), split 60/40 into ``C_boot`` and ``C_val``.""" + table = group_table(examples) + strata: dict[str, list[str]] = {} + for group, info in table.items(): + strata.setdefault(info.primary_family, []).append(group) + chosen: list[str] = [] + boot_groups: list[str] = [] + for family in sorted(strata): + ordered = _order(strata[family], seed, f"compile:{family}") + taken = ordered[: _take(ordered, fraction)] + chosen += taken + boot_groups += taken[: int(len(taken) * boot + 0.5)] + return CompileSplit(frozenset(chosen), frozenset(boot_groups), frozenset(chosen) - frozenset(boot_groups)) + + +def outer_folds(examples: Sequence[Example], *, k: int = OUTER_K, seed: int = 0, exclude: Iterable[str] = ()) -> list[Fold]: + """Grouped K-fold over the groups outside ``exclude``, stratified by (primary family, label mix), one repetition.""" + excluded = set(exclude) + table = {g: i for g, i in group_table(examples).items() if g not in excluded} + strata: dict[tuple[str, tuple[int, ...]], list[str]] = {} + for group, info in table.items(): + strata.setdefault((info.primary_family, tuple(sorted(info.labels))), []).append(group) + assigned: list[set[str]] = [set() for _ in range(k)] + turn = 0 + for key in sorted(strata): + for group in _order(strata[key], seed, f"outer:{key}"): + assigned[turn % k].add(group) + turn += 1 + return [Fold("outer", f"outer-{i}", frozenset(groups)) for i, groups in enumerate(assigned)] + + +def leave_one_source_out(examples: Sequence[Example], *, exclude: Iterable[str] = ()) -> list[Fold]: + excluded = set(exclude) + by_source: dict[str, set[str]] = {} + for example in examples: + if example.group not in excluded: + by_source.setdefault(example.source, set()).add(example.group) + return [Fold("leave_one_source_out", f"source-{s}", frozenset(g), held_source=s) for s, g in sorted(by_source.items())] + + +def leave_one_framework_out( + examples: Sequence[Example], *, exclude: Iterable[str] = (), min_groups: int = FRAMEWORK_MIN_GROUPS +) -> list[Fold]: + """One fold per framework seen; below ``min_groups`` groups it is ``insufficient data`` and evaluates nothing.""" + excluded = set(exclude) + by_framework: dict[str, set[str]] = {} + for example in examples: + for framework in example.frameworks: + if example.group not in excluded: + by_framework.setdefault(framework, set()).add(example.group) + folds = [] + for framework, groups in sorted(by_framework.items()): + enough = len(groups) >= min_groups + folds.append( + Fold( + "leave_one_framework_out", f"framework-{framework}", frozenset(groups) if enough else frozenset(), + held_framework=framework, status="ok" if enough else INSUFFICIENT, + ) + ) # fmt: skip + return folds + + +def assert_disjoint(split: CompileSplit, folds: Iterable[Fold]) -> None: + """The compile split shares no group with any evaluation fold; outer folds are pairwise disjoint.""" + seen: dict[str, str] = {} + for fold in folds: + shared = split.groups & fold.eval_groups + if shared: + raise FoldError(f"{fold.name}: groups in both the compile split and the evaluation fold: {sorted(shared)[:5]}") + if fold.kind == "outer": + for group in fold.eval_groups: + if group in seen: + raise FoldError(f"group {group!r} in {seen[group]} and {fold.name}") + seen[group] = fold.name + + +def folds_digest(split: CompileSplit, folds: Iterable[Fold]) -> str: + canonical = { + "compile": sorted(split.groups), "boot": sorted(split.boot), + "folds": [[f.kind, f.name, sorted(f.eval_groups), f.held_source, f.held_framework, f.status] for f in folds], + } # fmt: skip + return hashlib.sha256(json.dumps(canonical, sort_keys=True).encode("utf-8")).hexdigest() + + +__all__ = [ + "COMPILE_FRACTION", "FRAMEWORK_MIN_GROUPS", "INSUFFICIENT", "OUTER_K", "CompileSplit", "Fold", "FoldError", "GroupInfo", + "assert_disjoint", "compile_split", "folds_digest", "group_table", "leave_one_framework_out", "leave_one_source_out", + "outer_folds", +] # fmt: skip diff --git a/src/openultrasast/learn/retrieve.py b/src/openultrasast/learn/retrieve.py new file mode 100644 index 0000000..928fd4b --- /dev/null +++ b/src/openultrasast/learn/retrieve.py @@ -0,0 +1,277 @@ +"""Retrieval from the local memory, and the evaluation boundary (learned-decision-engine design 4.2, Req 3.2, 3.4). + +No model. Stage 1 ranks the eligible examples by a **Gower distance** over the allow-listed features of the active +profile, computed per instrument block so that missing is explicit, never imputed: + +- both blocks ``ran``: the mean over the block's features of ``|a - b| / cap`` (ints), ``|a - b|`` (fractions; + the one unbounded mean, ``verify.turns_mean``, over a cap of 10), ``0``/``1`` for bools and enums; ``null`` in + both is ``0``, in one is ``1``; +- different states (``ran`` vs ``none``/``failed``/``not_applicable``, or ``failed`` vs ``none``): ``1``; the + same non-``ran`` state: ``0``; +- ``D`` is the uniform mean over the profile's blocks; prior features are dropped with ``priors = "off"``. + +Ties break by ``sha256(example id + seed)``; the first ``n1 = 40`` go to stage 2, a cosine re-rank of the code +embeddings, ``r = lam (1 - D) + (1 - lam) cos``. Without the target's vector the whole retrieval is +``signals_only``; an example without a vector scores ``cos = 1 - D`` (its signal similarity, so it keeps its +stage-1 standing). The **balance rule** then takes ``k_ret = 6``: at most ``k/2`` per label, at most 2 per +repository group, at most 2 from other families while the candidate's family has examples left; ``short`` when +the pool cannot fill it. + +**The boundary.** :func:`eligible` is the only way an example enters a prompt -- retrieved examples and compiled +demonstrations alike: not the target's group, not a group the fold evaluates, not the held-out source, not the +held-out framework, and (outside deployment) not a near-duplicate of the target (cosine >= 0.98). The program +repeats the check on the ids it actually rendered (:func:`assert_boundary`). +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from typing import Any + +from .embeddings import cosine +from .examples import Example +from .folds import Fold +from .labels import repo_name +from .schema import UNBOUNDED_FLOATS, FeatureSpec, features_for, instruments_for + +N1 = 40 +K_RET = 6 +LAM = 0.5 +NEAR_DUPLICATE_COS = 0.98 +UNBOUNDED_CAP = 10.0 +PER_GROUP = 2 + + +class BoundaryViolation(AssertionError): + """An example or demonstration outside the evaluation boundary reached a prompt.""" + + +@dataclass(frozen=True) +class Target: + """What retrieval knows of a candidate: its group (for the boundary), family, record and excerpt vector.""" + + group: str + family: str + profile: str + x: Mapping[str, Any] + instruments: Mapping[str, Mapping[str, Any]] + vector: Sequence[float] | None = None + + @classmethod + def of(cls, example: Example, vector: Sequence[float] | None = None) -> Target: + return cls(example.group, example.family, example.profile, example.x, example.instruments, vector) + + +def _blocks(profile: str, priors: str) -> list[tuple[str, list[FeatureSpec]]]: + specs = [s for s in features_for(profile) if not (s.prior and priors == "off")] + blocks = [(name, [s for s in specs if s.instrument == name]) for name in instruments_for(profile)] + return [(name, block) for name, block in blocks if block] + + +def _feature_distance(spec: FeatureSpec, a: Any, b: Any) -> float: + if a is None and b is None: + return 0.0 + if a is None or b is None: + return 1.0 + if spec.type == "int": + return min(1.0, abs(int(a) - int(b)) / float(spec.cap or 1)) + if spec.type == "float": + cap = UNBOUNDED_CAP if spec.name in UNBOUNDED_FLOATS else 1.0 + return min(1.0, abs(float(a) - float(b)) / cap) + return 0.0 if a == b else 1.0 + + +def gower( + a_x: Mapping[str, Any], + a_instruments: Mapping[str, Mapping[str, Any]], + b_x: Mapping[str, Any], + b_instruments: Mapping[str, Mapping[str, Any]], + profile: str, + *, + priors: str = "off", +) -> float: + """The per-instrument Gower distance of two records of ``profile`` (module docstring), in [0, 1].""" + total, blocks = 0.0, _blocks(profile, priors) + for name, specs in blocks: + sa, sb = a_instruments.get(name, {}).get("state", "none"), b_instruments.get(name, {}).get("state", "none") + if sa != sb: + total += 1.0 + elif sa == "ran": + total += sum(_feature_distance(s, a_x.get(s.name), b_x.get(s.name)) for s in specs) / len(specs) + return total / len(blocks) if blocks else 0.0 + + +def ineligible(example: Example, target: Target, fold: Fold, *, cos: float | None = None) -> str | None: + """Why ``example`` may not enter ``target``'s prompt under ``fold``, or ``None`` when it may.""" + if example.group and example.group == target.group: + return "target group" + if example.group in fold.eval_groups: + return "evaluated group" + if fold.held_source is not None and example.source == fold.held_source: + return "held-out source" + if fold.held_framework is not None and fold.held_framework in example.frameworks: + return "held-out framework" + if fold.phase != "deployment" and cos is not None and cos >= NEAR_DUPLICATE_COS: + return "near duplicate" + return None + + +def eligible(example: Example, target: Target, fold: Fold, *, cos: float | None = None) -> bool: + return ineligible(example, target, fold, cos=cos) is None + + +@dataclass(frozen=True) +class Retrieval: + examples: tuple[Example, ...] + mode: str # signals+embeddings | signals_only + balance: str # ok | short + near_duplicates: int + pool: int # eligible examples before stage 1 + without_vector: int + + @property + def neighbours(self) -> dict[str, dict[str, int]]: + """Counts by label and family (never identities), for the decision row.""" + labels: dict[str, int] = {} + families: dict[str, int] = {} + for example in self.examples: + labels[str(example.label)] = labels.get(str(example.label), 0) + 1 + families[example.family] = families.get(example.family, 0) + 1 + return {"label": dict(sorted(labels.items())), "family": dict(sorted(families.items()))} + + +def _tie(example_id: str, seed: int) -> str: + return hashlib.sha256(f"{example_id}{seed}".encode()).hexdigest() + + +def balance(ranked: Sequence[Example], family: str, k: int = K_RET) -> tuple[list[Example], str]: + """Take up to ``k`` in rank order: <= k/2 per label, <= 2 per group, <= 2 other families while the target's family + has examples left; then, if short, fill from other families under the label and group caps.""" + per_label, other_cap = k // 2, min(2, k) + taken: list[Example] = [] + labels: dict[int, int] = {} + groups: dict[str, int] = {} + + def fits(example: Example) -> bool: + return labels.get(example.label, 0) < per_label and groups.get(example.group, 0) < PER_GROUP + + def take(example: Example) -> None: + taken.append(example) + labels[example.label] = labels.get(example.label, 0) + 1 + groups[example.group] = groups.get(example.group, 0) + 1 + + others = 0 + for example in ranked: + if len(taken) == k: + break + if example.family != family and others >= other_cap: + continue + if fits(example): + others += example.family != family + take(example) + for example in ranked: + if len(taken) == k: + break + if example.id not in {e.id for e in taken} and fits(example): + take(example) + pool_labels = {e.label for e in ranked} + short = len(taken) < k or any(labels.get(lab, 0) < min(2, per_label) for lab in (0, 1) if lab in pool_labels) + return taken, "short" if short else "ok" + + +def retrieve( + target: Target, + pool: Iterable[Example], + fold: Fold, + *, + vectors: Mapping[str, Sequence[float]] | None = None, + n1: int = N1, + k: int = K_RET, + lam: float = LAM, + seed: int = 0, + priors: str = "off", +) -> Retrieval: + """The ``k`` balanced examples for ``target`` from the eligible part of ``pool`` (module docstring).""" + candidates = [e for e in pool if e.profile == target.profile and eligible(e, target, fold)] + scored = sorted( + ((gower(target.x, target.instruments, e.x, e.instruments, target.profile, priors=priors), e) for e in candidates), + key=lambda pair: (pair[0], _tie(pair[1].id, seed)), + )[:n1] + mode = "signals+embeddings" if target.vector is not None and vectors else "signals_only" + near = without = 0 + reranked: list[tuple[float, str, Example]] = [] + for distance, example in scored: + similarity = 1.0 - distance + vector = (vectors or {}).get(example.excerpt_sha) if mode != "signals_only" else None + if mode != "signals_only" and target.vector is not None and vector is not None: + cos = cosine(target.vector, vector) + if not eligible(example, target, fold, cos=cos): + near += 1 + continue + score = lam * similarity + (1 - lam) * cos + else: + without += mode != "signals_only" + score = similarity + reranked.append((-score, _tie(example.id, seed), example)) + ranked = [example for _, _, example in sorted(reranked, key=lambda t: (t[0], t[1]))] + chosen, state = balance(ranked, target.family, k) + return Retrieval(tuple(chosen), mode, state, near, len(candidates), without) + + +def assert_boundary( + ids: Iterable[str], examples: Mapping[str, Example], target: Target, fold: Fold, *, vectors: Mapping[str, Sequence[float]] | None = None +) -> None: + """Raise :class:`BoundaryViolation` when any rendered id is unknown or ineligible for ``target`` under ``fold``.""" + for example_id in ids: + example = examples.get(example_id) + if example is None: + raise BoundaryViolation(f"rendered example {example_id[:12]} is not in the memory snapshot") + vector = (vectors or {}).get(example.excerpt_sha) + cos = cosine(target.vector, vector) if target.vector is not None and vector is not None else None + why = ineligible(example, target, fold, cos=cos) + if why is not None: + raise BoundaryViolation(f"rendered example {example_id[:12]} is outside the boundary of {fold.name}: {why}") + + +def demonstrations_for( + demos: Sequence[Mapping[str, Any]], examples: Mapping[str, Example], target: Target, fold: Fold +) -> list[Mapping[str, Any]]: + """The compiled demonstrations usable for ``target``: an ineligible one is swapped for its alternate when that + is eligible, and dropped otherwise (leave-one-framework-out on a demonstration's framework).""" + out: list[Mapping[str, Any]] = [] + for demo in demos: + example = examples.get(str(demo["example_id"])) + if example is not None and eligible(example, target, fold): + out.append(demo) + continue + alternate = demo.get("alternate") + alt = examples.get(str(alternate.get("example_id"))) if isinstance(alternate, Mapping) else None + if isinstance(alternate, Mapping) and alt is not None and eligible(alt, target, fold): + out.append(alternate) + return out + + +def deployment_fold(repository: str, groups: Iterable[str]) -> tuple[Fold, str | None]: + """The boundary of a deployed scan: a repository that is a corpus group excludes its own cases.""" + name = repo_name(repository) if repository else "" + if name and name in set(groups): + return Fold( + "deployment", "deployment", frozenset({name}), phase="deployment" + ), "repository is in the corpus: its own cases excluded" + return Fold("deployment", "deployment", frozenset(), phase="deployment"), None + + +def seed_memory(examples: Iterable[Example], case_group: str, fold: Fold | None = None) -> list[Example]: + """The examples a plane Run case may be seeded with: none of its own group (nor of the fold's evaluated groups, + held source or framework). The host filters; the sandbox never sees the rest.""" + target = Target(case_group, "", "", {}, {}) + boundary = fold or Fold("deployment", "seed", frozenset(), phase="deployment") + return [e for e in examples if e.group != case_group and eligible(e, target, boundary)] + + +__all__ = [ + "K_RET", "LAM", "N1", "NEAR_DUPLICATE_COS", "BoundaryViolation", "Retrieval", "Target", "assert_boundary", "balance", + "demonstrations_for", "deployment_fold", "eligible", "gower", "ineligible", "retrieve", "seed_memory", +] # fmt: skip diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index 48f337a..b9d6abd 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -73,6 +73,10 @@ SPEC_OWNED = { "learned-decision-engine Req 8.1: model role classification per file chunk (model_sinks.py moved in); " "its Run template comes with the harvest (task 5)" ), + "learn.retrieve": ( + "learned-decision-engine Req 3.2/3.4: retrieval from the local memory and the evaluation boundary; " + "the program (task 6.4) and `ousast learn compile` (6.5) are its callers" + ), "plane.workspaces": "ai-service-plane task 4: Workspace manifests per case pin of a population; `ousast plane workspaces`", "plane.runner": "ai-service-plane Req 4.1-4.4: the ax-task-runner entrypoint (PID 1 of the task image); delivers the output directory", } diff --git a/tests/learn_fixtures.py b/tests/learn_fixtures.py new file mode 100644 index 0000000..31f1a6f --- /dev/null +++ b/tests/learn_fixtures.py @@ -0,0 +1,64 @@ +"""Shared fixtures for the decision-engine tests: example rows with valid feature records, and scripted clients.""" + +from __future__ import annotations + +import hashlib +from collections.abc import Mapping, Sequence +from typing import Any + +from openultrasast.learn.examples import Example +from openultrasast.learn.features import NOT_APPLICABLE, Part, engine_part, quick_part, record + + +def x_record(*, hits: int = 0, engine: str = "ran", findings: int = 0, language: str = "python", prior_hits: int = 0) -> dict[str, Any]: + """A valid ``static`` record: ``hits`` enabled quick hits, the engine in state ``engine`` with ``findings``.""" + quick = quick_part([("r", "enabled")] * hits, {}) + quick = Part("ran", {**quick.values, "qr.prior_hits": prior_hits}, None) + parts: dict[str, Part] = {"quick": quick, "delta": NOT_APPLICABLE, "facts": NOT_APPLICABLE} + if engine == "ran": + parts["engine"] = engine_part([{"rung": "suspicion"}] * findings, completion=1.0, degraded=False) + elif engine != "none": + parts["engine"] = Part(engine) + return record("a.py::f", "injection", language, "static", parts) + + +def example( + name: str, + *, + group: str, + label: int, + family: str = "injection", + source: str = "pairs", + frameworks: Sequence[str] = (), + hits: int = 0, + engine: str = "ran", + findings: int = 0, + license: str = "", +) -> Example: + rec = x_record(hits=hits, engine=engine, findings=findings) + return Example( + id=hashlib.sha256(name.encode()).hexdigest(), group=group, source=source, frameworks=tuple(frameworks), family=family, label=label, + language="python", profile="static", unit="pin", x=rec["x"], instruments=rec["instruments"], roles=(), + excerpt_sha=hashlib.sha256(f"excerpt:{name}".encode()).hexdigest(), license=license, + ) # fmt: skip + + +def corpus(groups: int = 12, per_group: int = 4) -> list[Example]: + """``groups`` repository groups over three sources and two frameworks, each with positives and negatives.""" + out: list[Example] = [] + for g in range(groups): + source = ("pairs", "population-v1", "population-v2")[g % 3] + frameworks = ("flask",) if g % 4 == 0 else ("django",) if g % 4 == 1 else () + family = ("injection", "path")[g % 2] + for i in range(per_group): + out.append( + example( + f"g{g}-e{i}", group=f"org{g}/repo{g}", label=i % 2, family=family, source=source, frameworks=frameworks, + hits=(g + i) % 5, findings=i % 3, + ) + ) # fmt: skip + return out + + +def by_id(examples: Sequence[Example]) -> Mapping[str, Example]: + return {e.id: e for e in examples} diff --git a/tests/test_learn_folds.py b/tests/test_learn_folds.py new file mode 100644 index 0000000..6557d7a --- /dev/null +++ b/tests/test_learn_folds.py @@ -0,0 +1,76 @@ +"""Folds by repository group (learned-decision-engine design 4.3-4.4, task 6.3).""" + +from __future__ import annotations + +import pytest +from learn_fixtures import corpus + +from openultrasast.learn.folds import ( + INSUFFICIENT, + Fold, + FoldError, + assert_disjoint, + compile_split, + folds_digest, + leave_one_framework_out, + leave_one_source_out, + outer_folds, +) + + +def test_compile_split_is_a_quarter_of_groups_and_disjoint_from_every_evaluation_fold() -> None: + examples = corpus(groups=20) + split = compile_split(examples, seed=1) + groups = {e.group for e in examples} + assert len(split.groups) == 6 and split.groups <= groups # 25% of each family's 10 groups, rounded: 3 + 3 + assert split.boot | split.val == split.groups and not split.boot & split.val and len(split.boot) == 4 + folds = [ + *outer_folds(examples, seed=1, exclude=split.groups), + *leave_one_source_out(examples, exclude=split.groups), + *leave_one_framework_out(examples, exclude=split.groups, min_groups=2), + ] + assert_disjoint(split, folds) + for fold in folds: + assert not fold.eval_groups & split.groups + + +def test_outer_folds_partition_the_remaining_groups_stratified() -> None: + examples = corpus(groups=20) + split = compile_split(examples, seed=1) + folds = outer_folds(examples, seed=1, exclude=split.groups) + assert len(folds) == 5 + union = set().union(*(f.eval_groups for f in folds)) + assert union == {e.group for e in examples} - split.groups and sum(len(f.eval_groups) for f in folds) == len(union) + assert max(len(f.eval_groups) for f in folds) - min(len(f.eval_groups) for f in folds) <= 1 + + +def test_folds_are_deterministic_by_seed() -> None: + examples = corpus(groups=20) + a, b = compile_split(examples, seed=7), compile_split(examples, seed=7) + assert a == b and outer_folds(examples, seed=7) == outer_folds(examples, seed=7) + assert folds_digest(a, outer_folds(examples, seed=7)) == folds_digest(b, outer_folds(examples, seed=7)) + assert any(compile_split(examples, seed=s).groups != a.groups for s in range(1, 6)) + + +def test_leave_one_source_and_framework_out() -> None: + examples = corpus(groups=12) + sources = leave_one_source_out(examples) + assert {f.held_source for f in sources} == {"pairs", "population-v1", "population-v2"} + for fold in sources: + assert fold.eval_groups == {e.group for e in examples if e.source == fold.held_source} + frameworks = {f.held_framework: f for f in leave_one_framework_out(examples)} + assert set(frameworks) == {"flask", "django"} + assert all(f.status == INSUFFICIENT and not f.eval_groups for f in frameworks.values()) # 3 groups each < 10 + enough = {f.held_framework: f for f in leave_one_framework_out(examples, min_groups=3)} + assert enough["flask"].status == "ok" and enough["flask"].eval_groups == {e.group for e in examples if "flask" in e.frameworks} + + +def test_a_shared_group_is_refused() -> None: + examples = corpus(groups=8) + split = compile_split(examples, seed=0) + leaked = Fold("outer", "outer-x", frozenset({next(iter(split.groups))})) + with pytest.raises(FoldError, match="compile split"): + assert_disjoint(split, [leaked]) + twice = [Fold("outer", "a", frozenset({"g"})), Fold("outer", "b", frozenset({"g"}))] + with pytest.raises(FoldError, match="in a and b"): + assert_disjoint(split, twice) diff --git a/tests/test_learn_retrieve.py b/tests/test_learn_retrieve.py new file mode 100644 index 0000000..de2827e --- /dev/null +++ b/tests/test_learn_retrieve.py @@ -0,0 +1,175 @@ +"""Retrieval and the evaluation boundary (learned-decision-engine design 4.2, task 6.3): Gower distance with explicit +missing, the embedding re-rank, the balance rule, and ``test_boundary``.""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest +from learn_fixtures import by_id, corpus, example, x_record + +from openultrasast.learn.folds import compile_split, leave_one_framework_out, leave_one_source_out, outer_folds +from openultrasast.learn.retrieve import ( + BoundaryViolation, + Target, + assert_boundary, + balance, + demonstrations_for, + deployment_fold, + eligible, + gower, + retrieve, + seed_memory, +) +from openultrasast.learn.schema import features_for, instruments_for + +BLOCKS = len(instruments_for("static")) +ENGINE = [s for s in features_for("static") if s.instrument == "engine"] +QUICK_NO_PRIOR = [s for s in features_for("static") if s.instrument == "quick" and not s.prior] +QUICK_ALL = [s for s in features_for("static") if s.instrument == "quick"] + + +def _d(a: dict, b: dict, **kw: str) -> float: + return gower(a["x"], a["instruments"], b["x"], b["instruments"], "static", **kw) + + +def test_gower_on_hand_computed_fixtures() -> None: + base = x_record(hits=2, findings=1) + assert _d(base, base) == 0.0 + assert _d(base, x_record(hits=2, engine="none")) == pytest.approx(1 / BLOCKS) # ran vs none: the block is 1 + assert _d(x_record(engine="none"), x_record(engine="none")) == 0.0 # the same non-ran state is 0 + assert _d(x_record(engine="failed"), x_record(engine="none")) == pytest.approx(1 / BLOCKS) # failed vs none differ + # 0 vs 10 enabled hits of cap 20 is 0.5; the mechanism bucket `other` (unknown rules) moves 10 of cap 10 + expected = (10 / 20 + 10 / 10) / len(QUICK_NO_PRIOR) / BLOCKS + assert _d(x_record(hits=0), x_record(hits=10)) == pytest.approx(expected) + # null in one, a value in the other: rung_max is null without findings, `suspicion` with one + zero, one = x_record(findings=0), x_record(findings=1) + assert _d(zero, one) == pytest.approx((1 / 10 + 1) / len(ENGINE) / BLOCKS) # findings 0 vs 1 (cap 10), rung null vs set + + +def test_prior_features_are_dropped_with_priors_off() -> None: + plain, prior = x_record(prior_hits=0), x_record(prior_hits=5) + assert _d(plain, prior) == 0.0 + assert _d(plain, prior, priors="on") == pytest.approx(5 / 10 / len(QUICK_ALL) / BLOCKS) + + +def _target(**kw: object) -> Target: + rec = x_record(**kw) # type: ignore[arg-type] + return Target("target/repo", "injection", "static", rec["x"], rec["instruments"]) + + +def test_rerank_by_embeddings_with_lam_and_signals_only_without_vectors() -> None: + near_signal = example("near-signal", group="a/a", label=1, hits=0) + near_code = example("near-code", group="b/b", label=0, hits=8) + vectors = {near_signal.excerpt_sha: [0.0, 1.0], near_code.excerpt_sha: [1.0, 0.5]} + fold = outer_folds([near_signal, near_code], k=1)[0] + fold = replace(fold, eval_groups=frozenset()) + target = replace(_target(hits=0), vector=[1.0, 0.0]) + by_code = retrieve(target, [near_signal, near_code], fold, vectors=vectors, lam=0.0, k=2) + assert [e.id for e in by_code.examples][0] == near_code.id and by_code.mode == "signals+embeddings" + by_signal = retrieve(replace(target, vector=None), [near_signal, near_code], fold, vectors=vectors, lam=0.5, k=2) + assert by_signal.mode == "signals_only" and by_signal.examples[0].id == near_signal.id + mixed = retrieve(target, [near_signal, near_code], fold, vectors=vectors, lam=0.5, k=2) + assert mixed.mode == "signals+embeddings" and {e.id for e in mixed.examples} == {near_signal.id, near_code.id} + + +def test_balance_caps_labels_groups_and_other_families() -> None: + same = [example(f"s{i}", group=f"g{i % 2}/r", label=1) for i in range(6)] + taken, state = balance(same, "injection", 6) + assert len(taken) == 3 and state == "short" # <= k/2 positives, and no negatives in the pool + mixed = [example(f"m{i}", group=f"g{i}/r", label=i % 2, family="path" if i < 4 else "injection") for i in range(12)] + taken, state = balance(mixed, "injection", 6) + assert sum(e.family != "injection" for e in taken) <= 2 and state == "ok" + assert sum(e.label for e in taken) == 3 + crowded = [example(f"c{i}", group="one/repo", label=i % 2) for i in range(10)] + taken, state = balance(crowded, "injection", 6) + assert len(taken) == 2 and state == "short" # at most 2 per repository group + + +def test_near_duplicates_drop_in_evaluation_and_stay_in_deployment() -> None: + twin = example("twin", group="fork/repo", label=1) + other = example("other", group="x/y", label=0) + vectors = {twin.excerpt_sha: [1.0, 0.0], other.excerpt_sha: [0.0, 1.0]} + target = replace(_target(), vector=[1.0, 0.0]) + evaluation = replace(outer_folds([twin], k=1)[0], eval_groups=frozenset()) + got = retrieve(target, [twin, other], evaluation, vectors=vectors, k=2) + assert got.near_duplicates == 1 and [e.id for e in got.examples] == [other.id] + deployed, _ = deployment_fold("https://github.com/someone/else", ["fork/repo"]) + kept = retrieve(target, [twin, other], deployed, vectors=vectors, k=2) + assert kept.near_duplicates == 0 and twin.id in {e.id for e in kept.examples} + + +def test_deployment_excludes_the_scanned_repository_when_it_is_a_corpus_group() -> None: + own = example("own", group="acme/webapp", label=1) + fold, note = deployment_fold("https://github.com/Acme/WebApp.git", ["acme/webapp", "x/y"]) + assert note == "repository is in the corpus: its own cases excluded" + assert not eligible(own, Target("someone/else", "injection", "static", own.x, own.instruments), fold) + assert deployment_fold("https://github.com/new/repo", ["acme/webapp"])[1] is None + + +def _all_folds(examples: list) -> list: + split = compile_split(examples, seed=3) + return [ + *outer_folds(examples, seed=3, exclude=split.groups), + *leave_one_source_out(examples, exclude=split.groups), + *[f for f in leave_one_framework_out(examples, exclude=split.groups, min_groups=2) if f.status == "ok"], + ] + + +def test_boundary() -> None: + """For outer, leave-one-source-out and leave-one-framework-out folds, every retrieved example and every + demonstration that reaches a prompt is eligible; ``seed_memory`` writes no example of the case's group.""" + examples = corpus(groups=16) + index = by_id(examples) + split = compile_split(examples, seed=3) + demos = [ + {"example_id": e.id, "label": e.label, "alternate": {"example_id": alt.id, "label": alt.label}} + for e, alt in zip( + [e for e in examples if e.group in split.boot and e.frameworks][:3], + [e for e in examples if e.group in split.boot and not e.frameworks][:3], + strict=False, + ) + ] + folds = _all_folds(examples) + assert {f.kind for f in folds} == {"outer", "leave_one_source_out", "leave_one_framework_out"} + checked = 0 + for fold in folds: + for target_example in [e for e in examples if e.group in fold.eval_groups]: + target = Target.of(target_example) + got = retrieve(target, examples, fold, k=6) + usable = demonstrations_for(demos, index, target, fold) + rendered = [e.id for e in got.examples] + [str(d["example_id"]) for d in usable] + assert_boundary(rendered, index, target, fold) + for example_id in rendered: + chosen = index[example_id] + assert chosen.group != target.group and chosen.group not in fold.eval_groups + assert fold.held_source is None or chosen.source != fold.held_source + assert fold.held_framework is None or fold.held_framework not in chosen.frameworks + checked += 1 + assert checked > 100 + for group in {e.group for e in examples}: + assert all(e.group != group for e in seed_memory(examples, group)) + + +def test_a_demonstration_of_the_held_out_framework_is_swapped_for_its_alternate() -> None: + flask = example("demo-flask", group="d/flask", label=1, frameworks=("flask",)) + alt = example("demo-plain", group="d/plain", label=1) + target_example = example("t", group="t/t", label=0, frameworks=("flask",)) + target = Target.of(target_example) + fold = leave_one_framework_out([flask, target_example], min_groups=1)[0] + assert fold.held_framework == "flask" and target_example.group in fold.eval_groups + demos = [{"example_id": flask.id, "alternate": {"example_id": alt.id}}] + usable = demonstrations_for(demos, by_id([flask, alt]), target, fold) + assert [d["example_id"] for d in usable] == [alt.id] + assert demonstrations_for([{"example_id": flask.id, "alternate": None}], by_id([flask]), target, fold) == [] + + +def test_assert_boundary_catches_an_ineligible_rendered_id() -> None: + examples = corpus(groups=6) + fold = outer_folds(examples, k=3)[0] + target_example = next(e for e in examples if e.group in fold.eval_groups) + sibling = next(e for e in examples if e.group == target_example.group and e.id != target_example.id) + with pytest.raises(BoundaryViolation, match="target group"): + assert_boundary([sibling.id], by_id(examples), Target.of(target_example), fold) + with pytest.raises(BoundaryViolation, match="not in the memory"): + assert_boundary(["0" * 64], by_id(examples), Target.of(target_example), fold)