WO55: extension-point registries (gates, stage types, volume providers, license policy)

src/pipeline/contracts.py

@@ -20,9 +20,21 @@
 
 TaskKind = Literal["diacritization", "g2p", "transliteration"]
 METRIC_KEYS = ("cer_max", "wer_max", "per_max")
-KNOWN_STAGES = frozenset(
-    {"sft", "oracle_relabel", "distill", "decode_policy", "plane", "lexicon_fusion"}
-)
+
+# Stage types are a REGISTRY (WO55): new techniques register themselves,
+# contracts.py never edits. Seeded with the proven set.
+_STAGE_TYPES: set[str] = set()
+
+
+def register_stage_type(name: str) -> None:
+    _STAGE_TYPES.add(name)
+
+
+for _s in ("sft", "oracle_relabel", "distill", "decode_policy", "plane",
+           "lexicon_fusion"):
+    register_stage_type(_s)
+
+KNOWN_STAGES = _STAGE_TYPES  # backwards-compatible alias (live set)
 LANG_RE = re.compile(r"^[a-z]{2,3}-[a-z0-9-]+$")  # recipes: <lang>-<task>
 CORPUS_ID_RE = re.compile(r"^[a-z][a-z0-9-]*$")  # corpora: source names
 CORPUS_REF_RE = re.compile(
@@ -48,7 +60,10 @@ class LicenseReview(_Strict):
 
 
 class CorpusSource(_Strict):
-    kind: Literal["hf-dataset", "bucket", "gh-release"]
+    # kind is free-form schema data; the planner's volume-provider
+    # registry decides what is mountable (typos fail at plan time with
+    # the registered kinds in the error)
+    kind: str
     id: str
     revision: str | None = None
     sha256: str | None = None
@@ -193,10 +208,27 @@ def _license_blocked(lic: str) -> bool:
     return lic == "UNVERIFIED" or lic.startswith("LicenseRef-")
 
 
+class LicensePolicy:
+    """The license gate's policy, encapsulated: varying it (stricter or
+    looser) is constructing a policy object, never editing the schema."""
+
+    def __init__(self, allow_waivers: bool = True) -> None:
+        self.allow_waivers = allow_waivers
+
+    def blocked(self, source: CorpusSource) -> bool:
+        if not _license_blocked(source.license):
+            return False
+        if not self.allow_waivers:
+            return True  # strict policy: restricted stays blocked regardless
+        review = source.license_review
+        return review is None or review.status != "owner-approved"
+
+
+DEFAULT_LICENSE_POLICY = LicensePolicy()
+
+
 def _license_blocked_source(s: CorpusSource) -> bool:
-    if not _license_blocked(s.license):
-        return False
-    return s.license_review is None or s.license_review.status != "owner-approved"
+    return DEFAULT_LICENSE_POLICY.blocked(s)
 
 
 def validate_recipe_dir(recipes_dir: Path, corpora_dir: Path) -> list[str]:

src/pipeline/gates.py

@@ -1,16 +1,23 @@
-"""WO52 M2 — machine-checkable gates.
+"""WO52 M2 gates — hardened by WO51, registried by WO55.
 
-Gates evaluate in the runtime's actual decode mode (WO51): the parity
-gate reads single-row ORT results, never a batched harness. A missing
-result fails the gate — silence is never success.
+Gates are REGISTERED objects: `evaluate()` runs every registered gate
+and adding a gate (conformance, regression, a future claims gate) is a
+registration, never an edit to this module. Each gate receives a
+GateContext; missing inputs fail the gate loudly — silence is never
+success.
 """
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
+from typing import Any, ClassVar
 
-from pipeline.contracts import RecipeSpec, load_corpus
+from pipeline.contracts import (
+    RecipeSpec,
+    _license_blocked_source,
+    load_corpus,
+)
 
 
 @dataclass
@@ -20,6 +27,37 @@ class GateResult:
     detail: str
 
 
+@dataclass
+class GateContext:
+    metrics: dict | None = None
+    parity: dict | None = None
+    corpora_dir: Path | None = None
+    conformance_path: Path | None = None
+    registry: dict | None = None
+    extra: dict[str, Any] = field(default_factory=dict)
+
+
+class BaseGate:
+    name: ClassVar[str] = "gate"
+
+    def check(self, recipe: RecipeSpec, ctx: GateContext) -> GateResult:
+        raise NotImplementedError
+
+
+_REGISTRY: dict[str, BaseGate] = {}
+
+
+def register_gate(gate: BaseGate) -> None:
+    _REGISTRY[gate.name] = gate
+
+
+def registered_gates() -> list[str]:
+    return sorted(_REGISTRY)
+
+
+# ---- the three proven gates -------------------------------------------------
+
+
 def check_quality(recipe: RecipeSpec, metrics: dict | None) -> GateResult:
     if metrics is None:
         return GateResult("quality", False, "no metrics provided")
@@ -73,18 +111,12 @@ def check_parity(recipe: RecipeSpec, parity: dict | None) -> GateResult:
 
 
 def check_license(recipe: RecipeSpec, corpora_dir: Path) -> GateResult:
-    from pipeline.contracts import _license_blocked_source
-
     names = {ref.split("@", 1)[0] for ref in recipe.corpora.values()}
     for name in sorted(names):
         try:
             corpus = load_corpus(Path(corpora_dir) / f"{name}.yaml")
         except Exception as e:  # noqa: BLE001 — restricted/unreadable registry entry
-            return GateResult(
-                "license",
-                False,
-                f"corpus '{name}': {e}",
-            )
+            return GateResult("license", False, f"corpus '{name}': {e}")
         bad = [s for s in corpus.sources if _license_blocked_source(s)]
         if bad:
             ids = ", ".join(f"{s.id}={s.license}" for s in bad)
@@ -97,14 +129,122 @@ def check_license(recipe: RecipeSpec, corpora_dir: Path) -> GateResult:
     return GateResult("license", True, f"corpora {sorted(names)} licensed")
 
 
+# ---- gate adapters (registry form of the functions above) -------------------
+
+
+class QualityGate(BaseGate):
+    name = "quality"
+
+    def check(self, recipe: RecipeSpec, ctx: GateContext) -> GateResult:
+        return check_quality(recipe, ctx.metrics)
+
+
+class ParityGate(BaseGate):
+    name = "parity"
+
+    def check(self, recipe: RecipeSpec, ctx: GateContext) -> GateResult:
+        return check_parity(recipe, ctx.parity)
+
+
+class LicenseGate(BaseGate):
+    name = "license"
+
+    def check(self, recipe: RecipeSpec, ctx: GateContext) -> GateResult:
+        return check_license(recipe, ctx.corpora_dir)
+
+
+# ---- WO55 gates (exist because the registry makes them cheap) ---------------
+
+
+class ConformanceGate(BaseGate):
+    """The frozen-fixture gate: a model ships conformance vectors or it
+    does not pass. Validated against the vectors schema; byte-exact
+    execution stays with `isxml conformance` (needs the artifact)."""
+
+    name = "conformance"
+
+    def check(self, recipe: RecipeSpec, ctx: GateContext) -> GateResult:
+        from pipeline.conformance import validate_vectors
+
+        if ctx.conformance_path is None:
+            return GateResult(
+                "conformance",
+                False,
+                "no conformance vectors file — a model ships frozen "
+                "fixtures or not at all",
+            )
+        path = Path(ctx.conformance_path)
+        if not path.is_file():
+            return GateResult("conformance", False,
+                              f"vectors file missing: {path}")
+        import json
+
+        vectors = json.loads(path.read_text(encoding="utf-8"))
+        problems = validate_vectors(vectors)
+        if problems:
+            return GateResult("conformance", False, "; ".join(problems))
+        return GateResult(
+            "conformance",
+            True,
+            f"{len(vectors['vectors'])} vectors, "
+            f"{vectors['decode']['mode']}",
+        )
+
+
+class RegressionGate(BaseGate):
+    """Flag-only comparison against the registry's best-known numbers:
+    a candidate worse than the shipped line never blocks silently — it
+    flags for the owner to waive (spec L2)."""
+
+    name = "regression"
+
+    def check(self, recipe: RecipeSpec, ctx: GateContext) -> GateResult:
+        if not ctx.registry or recipe.id not in ctx.registry:
+            return GateResult("regression", True,
+                              "no registry baseline for this recipe")
+        entry = ctx.registry[recipe.id]
+        best = (entry.get("best") or {}) if isinstance(entry, dict) else {}
+        flags: list[str] = []
+        improves: list[str] = []
+        for key in recipe.gates.quality:
+            metric = key.removesuffix("_max")
+            known = (best.get(metric))
+            value = (ctx.metrics or {}).get(metric)
+            if known is None or value is None:
+                continue
+            if value > known:
+                flags.append(f"{metric} {value} > best-known {known}")
+            elif value < known:
+                improves.append(f"{metric} {value} < best-known {known}")
+        if flags:
+            return GateResult("regression", True,
+                              "FLAG (owner waives): " + "; ".join(flags))
+        if improves:
+            return GateResult("regression", True, "improves: "
+                              + "; ".join(improves))
+        return GateResult("regression", True, "at parity with registry")
+
+
+register_gate(QualityGate())
+register_gate(ParityGate())
+register_gate(LicenseGate())
+register_gate(ConformanceGate())
+register_gate(RegressionGate())
+
+
 def evaluate(
     recipe: RecipeSpec,
     corpora_dir: Path,
     metrics: dict | None = None,
     parity: dict | None = None,
+    conformance_path: Path | None = None,
+    registry: dict | None = None,
 ) -> list[GateResult]:
-    return [
-        check_quality(recipe, metrics),
-        check_parity(recipe, parity),
-        check_license(recipe, corpora_dir),
-    ]
+    ctx = GateContext(
+        metrics=metrics,
+        parity=parity,
+        corpora_dir=corpora_dir,
+        conformance_path=conformance_path,
+        registry=registry,
+    )
+    return [gate.check(recipe, ctx) for gate in _REGISTRY.values()]

src/pipeline/planner.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import Any
 
 from pipeline.contracts import RecipeSpec, StageSpec, load_corpus
 from pipeline.runner import JobSpec, Volume
@@ -19,6 +20,32 @@
 REPO_ROOT = Path(__file__).resolve().parents[2]
 DEFAULT_CORPORA_DIR = REPO_ROOT / "corpora"
 
+# Volume providers are a REGISTRY (WO55): a new corpus source kind is a
+# registered mount strategy, never an edit to plan_stage. A provider
+# returns a Volume or None (skip).
+_VOLUME_PROVIDERS: dict[str, Any] = {}
+
+
+def register_volume_provider(kind: str, provider) -> None:
+    _VOLUME_PROVIDERS[kind] = provider
+
+
+def _bucket_volume(src, idx: int):
+    return Volume(f"hf://buckets/{src.id}", src.mount or f"/data/{idx}", "ro")
+
+
+def _dataset_volume(src, idx: int):
+    return Volume(f"hf://datasets/{src.id}", src.mount or f"/data/{idx}", "ro")
+
+
+def _skip_volume(src, idx: int):
+    return None  # release assets are fetched at runtime, not mounted
+
+
+register_volume_provider("bucket", _bucket_volume)
+register_volume_provider("hf-dataset", _dataset_volume)
+register_volume_provider("gh-release", _skip_volume)
+
 
 def plan_stage(
     recipe: RecipeSpec,
@@ -51,16 +78,18 @@ def plan_stage(
         seen.add(name)
         corpus = load_corpus(corpora_dir / f"{name}.yaml")
         for src in corpus.sources:
-            if src.kind == "gh-release":
-                continue  # release assets are fetched at runtime, not mounted
-            url = (
-                f"hf://buckets/{src.id}"
-                if src.kind == "bucket"
-                else f"hf://datasets/{src.id}"
-            )
-            mount = src.mount or f"/data/{idx}"
-            volumes.append(Volume(url, mount, "ro"))
-            idx += 1
+            provider = _VOLUME_PROVIDERS.get(src.kind)
+            if provider is None:
+                raise ValueError(
+                    f"corpus '{corpus.id}': no volume provider for source "
+                    f"kind '{src.kind}' (registered: "
+                    f"{sorted(_VOLUME_PROVIDERS)}) — register one via "
+                    "register_volume_provider"
+                )
+            vol = provider(src, idx)
+            if vol is not None:
+                volumes.append(vol)
+                idx += 1
     return JobSpec(
         name=run,
         command=command,

tests/test_pipeline_plan_gate.py

@@ -148,7 +148,15 @@ def test_gate_results_files_end_to_end(tmp_path: Path) -> None:
         parity=json.loads(parity.read_text()),
     )
     by_name = {g.name: g.passed for g in results}
-    assert by_name == {"quality": True, "parity": True, "license": True}
+    # the registry runs every registered gate; the conformance gate
+    # fails loudly on missing input and regression is flag-only
+    assert by_name == {
+        "quality": True,
+        "parity": True,
+        "license": True,
+        "conformance": False,
+        "regression": True,
+    }
 
 
 def test_plan_wraps_legacy_deps() -> None: