feat(v0.4-beta): learned-selector seam (model-free, shadow-only)

.gitignore

@@ -8,3 +8,6 @@
 # Python bytecode caches (eval/ + python-tests harnesses)
 __pycache__/
 *.pyc
+# v0.4-beta learned-selector: model/training code + weights stay LOCAL, never committed.
+# run_baselines.py discovers this via TOKENFOLD_LEARNED_MODULE and skips it cleanly when absent.
+/eval/learned/

eval/run_baselines.py

@@ -185,6 +185,53 @@ def sel_llmlingua_style(units: list[str], query: str) -> list[float]:
     "llmlingua_style": sel_llmlingua_style,
 }
 
+# Snapshot the deterministic set before any learned selector registers, so the report can label
+# which scores came from a model vs. a stdlib heuristic.
+DETERMINISTIC_SELECTORS = tuple(SELECTORS)
+
+
+# --- optional learned selector (v0.4-beta, model code lives locally, never committed) ----------
+# A trained keep/drop selector plugs in here as just another SELECTORS entry — no other harness
+# change. Its code + weights live in a *gitignored* local module ($TOKENFOLD_LEARNED_MODULE,
+# default `learned.selector`, under eval/) that owns the torch/transformers imports, so when the
+# ML stack or the local module is absent the import fails and the selector is cleanly skipped,
+# exactly like tiktoken and the tokenfold CLI above. Contract (model-research.md): the module
+# exposes LEARNED_SELECTORS: dict[str, callable], each callable (units, query) -> list[float]
+# emitting *scores only*; the harness's deterministic critical-atom forcing + ceiling allocator +
+# byte-copy assembly still own the output (so 100% critical-atom survival stays structural, and
+# `--gate` proves the learned selector cannot break it either).
+
+
+def _load_learned_selectors() -> dict:
+    import importlib
+
+    eval_dir = str(Path(__file__).resolve().parent)
+    if eval_dir not in sys.path:
+        sys.path.insert(0, eval_dir)  # importable whether run as a script or imported by a test
+    try:
+        mod = importlib.import_module(
+            os.environ.get("TOKENFOLD_LEARNED_MODULE", "learned.selector")
+        )
+        found = getattr(mod, "LEARNED_SELECTORS", {})
+    except Exception:  # ML absent / module absent / weights missing -> skip, never crash
+        return {}
+    learned = {k: v for k, v in found.items() if callable(v)} if isinstance(found, dict) else {}
+    # Integrity: a learned selector must never shadow a deterministic baseline. Overwriting one
+    # would run the model under that name yet label it deterministic (set-difference below) and
+    # silently drop the real baseline from the comparison. That defeats the whole report, so fail
+    # loud instead — a naming clash is a bug in the local module, not something to paper over.
+    clash = sorted(set(learned) & set(DETERMINISTIC_SELECTORS))
+    if clash:
+        raise ValueError(
+            f"learned selector name(s) {clash} shadow deterministic baselines; rename them in the "
+            "local model module — a learned selector must not overwrite a baseline."
+        )
+    return learned
+
+
+LEARNED_SELECTORS = _load_learned_selectors()
+SELECTORS.update(LEARNED_SELECTORS)
+
 
 # --- whole-pipeline compressor baselines ------------------------------------------------------
 # Unlike SELECTORS (which rank atomic units and get the harness's deterministic critical-atom
@@ -422,6 +469,9 @@ def build_report(fixtures: list[dict], ratios: list[float]) -> dict:
         "tokenizer": TOKENIZER,
         "fixture_count": len(fixtures),
         "selectors": list(SELECTORS),
+        "deterministic_selectors": list(DETERMINISTIC_SELECTORS),
+        # v0.4-beta: names registered from the gitignored local model module, [] when ML is absent.
+        "learned_selectors": [s for s in SELECTORS if s not in DETERMINISTIC_SELECTORS],
         "compressors": list(COMPRESSORS),
         "tokenfold_available": _TOKENFOLD_BIN is not None,
         "ratios": ratios,
@@ -482,10 +532,12 @@ def run_gate(fixtures: list[dict], ratios: list[float]) -> int:
 
 def _print_summary(report: dict) -> None:
     tf = "available" if report["tokenfold_available"] else "MISSING (skipped)"
+    learned = report.get("learned_selectors") or ["none (ML absent, skipped)"]
     print(
         f"# v0.4-alpha baselines  (tokenizer: {report['tokenizer']['backend']}, "
         f"deterministic-tokenfold: {tf})"
     )
+    print(f"# learned selectors: {', '.join(learned)}")
     print(f"# {report['fixture_count']} fixtures  ratios={report['ratios']}\n")
     print(f"{'baseline':<24}{'ratio':>6}{'task':>7}{'crit':>7}{'achieved':>10}{'over':>6}")
     for s in report["summary"]:

eval/test_learned_selector_wiring.py

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+"""Model-free wiring test for the v0.4-beta learned-selector seam in `run_baselines.py`.
+
+There is no ML here. It proves three things about the *plug-in mechanism* without any model
+installed, so the seam is verified now and the actual (gitignored) probe just has to satisfy the
+same `LEARNED_SELECTORS` contract later:
+
+  1. Absent module / ML  -> `_load_learned_selectors()` returns {} and never raises (clean skip,
+     like tiktoken and the tokenfold CLI).
+  2. A module that exposes `LEARNED_SELECTORS` gets its callables registered into `SELECTORS`.
+  3. Any registered score function still goes through the deterministic critical-atom forcing +
+     ceiling allocator, so 100% critical-atom survival and the token ceiling hold *by
+     construction* — a learned selector emits scores only, it cannot break the hard gates.
+
+Stdlib only, assert-based; run directly (`python eval/test_learned_selector_wiring.py`) or via
+pytest. No frameworks, no fixtures beyond the shipped task JSON.
+"""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))  # import run_baselines regardless of CWD
+import run_baselines as rb  # noqa: E402
+
+
+def test_absent_module_skips_cleanly():
+    # Point at a module that certainly does not exist: must return {} and not raise.
+    _set_env("TOKENFOLD_LEARNED_MODULE", "definitely_no_such_learned_module_xyz")
+    try:
+        assert rb._load_learned_selectors() == {}
+    finally:
+        _set_env("TOKENFOLD_LEARNED_MODULE", None)
+
+
+def test_module_with_contract_registers():
+    with tempfile.TemporaryDirectory() as d:
+        Path(d, "dummy_learned.py").write_text(
+            "LEARNED_SELECTORS = {"
+            "'learned_dummy': lambda units, query: [float(len(u)) for u in units]}\n",
+            encoding="utf-8",
+        )
+        sys.path.insert(0, d)
+        _set_env("TOKENFOLD_LEARNED_MODULE", "dummy_learned")
+        try:
+            loaded = rb._load_learned_selectors()
+            assert "learned_dummy" in loaded, loaded
+            assert callable(loaded["learned_dummy"])
+        finally:
+            _set_env("TOKENFOLD_LEARNED_MODULE", None)
+            sys.path.remove(d)
+
+
+def _write_dummy_module(dirpath, modname, body):
+    Path(dirpath, f"{modname}.py").write_text(body, encoding="utf-8")
+    sys.path.insert(0, dirpath)
+    _set_env("TOKENFOLD_LEARNED_MODULE", modname)
+
+
+def test_real_registration_path_flows_through_report():
+    """Exercise the ACTUAL register step run_baselines runs at import
+    (`SELECTORS.update(_load_learned_selectors())`) — not a hand-injection — and prove the loaded
+    selector reaches the measured registry, is labeled *learned* (never deterministic) in the
+    report, and still passes the hard gates. Breaking the real `.update()` line must fail here."""
+    with tempfile.TemporaryDirectory() as d:
+        _write_dummy_module(
+            d,
+            "dummy_flow",
+            "LEARNED_SELECTORS = {"
+            "'learned_flow': lambda units, query: [float(len(u)) for u in units]}\n",
+        )
+        rb.SELECTORS.update(rb._load_learned_selectors())  # the exact line run_baselines runs
+        try:
+            assert "learned_flow" in rb.SELECTORS
+            fixtures = rb.load_fixtures(Path(__file__).parent / "tasks" / "v04")
+            report = rb.build_report(fixtures, [0.25])
+            assert "learned_flow" in report["learned_selectors"], report["learned_selectors"]
+            assert "learned_flow" not in report["deterministic_selectors"]
+            assert rb.run_gate(fixtures, [0.25]) == 0  # learned selector must pass the hard gates
+        finally:
+            rb.SELECTORS.pop("learned_flow", None)
+            _set_env("TOKENFOLD_LEARNED_MODULE", None)
+            sys.path.remove(d)
+
+
+def test_collision_with_baseline_is_rejected():
+    """A learned selector that shadows a deterministic baseline name must fail loud, not silently
+    overwrite the baseline and get mislabeled as deterministic."""
+    with tempfile.TemporaryDirectory() as d:
+        _write_dummy_module(
+            d,
+            "dummy_clash",
+            "LEARNED_SELECTORS = {'bm25': lambda units, query: [0.0 for _ in units]}\n",
+        )
+        try:
+            raised = False
+            try:
+                rb._load_learned_selectors()
+            except ValueError:
+                raised = True
+            assert raised, "collision with deterministic baseline 'bm25' must raise"
+        finally:
+            _set_env("TOKENFOLD_LEARNED_MODULE", None)
+            sys.path.remove(d)
+
+
+def test_registered_selector_upholds_hard_gates():
+    """Inject an arbitrary learned score fn and drive a real fixture through run_one: critical
+    atoms must all survive and the ceiling must hold (unless the forced floor alone exceeds it) —
+    exactly the invariants `--gate` asserts, now proven for a non-deterministic selector too."""
+    fixtures = rb.load_fixtures(Path(__file__).parent / "tasks" / "v04")
+    assert fixtures, "no v04 fixtures found"
+    rb.SELECTORS["learned_dummy"] = lambda units, query: [float(len(u)) for u in units]
+    try:
+        for fx in fixtures:
+            res = rb.run_one(fx, "learned_dummy", 0.25)
+            assert res["critical_atom_survival"] == 1.0, res
+            ceiling_ok = (not res["over_budget"]) or (
+                res["forced_floor_tokens"] > res["budget_tokens"]
+            )
+            assert ceiling_ok, res
+    finally:
+        rb.SELECTORS.pop("learned_dummy", None)
+
+
+def _set_env(key, value):
+    import os
+
+    if value is None:
+        os.environ.pop(key, None)
+    else:
+        os.environ[key] = value
+
+
+if __name__ == "__main__":
+    test_absent_module_skips_cleanly()
+    test_module_with_contract_registers()
+    test_real_registration_path_flows_through_report()
+    test_collision_with_baseline_is_rejected()
+    test_registered_selector_upholds_hard_gates()
+    print("ok: learned-selector wiring holds (skip-when-absent, register+report flow, "
+          "baseline-collision rejected, gates-hold)")