From 6c1c5b44d750a512f2f3292e34073547673678b7 Mon Sep 17 00:00:00 2001 From: Ronald Tse Date: Mon, 19 Oct 2026 08:16:47 +0800 Subject: [PATCH] =?UTF-8?q?feat(pipeline):=20WO55=20=E2=80=94=20extension-?= =?UTF-8?q?point=20registries=20(OCP=20round)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The spine's extension points were closed lists; adding a gate, stage type, corpus source kind, or license policy meant editing core modules. Each is now a registry — closed for modification, open for extension: - gates: BaseGate + register_gate; evaluate() runs every registered gate with a GateContext (metrics/parity/corpora/conformance/registry as one encapsulated object). The proven three (quality, parity, license) are adapters over the unchanged check functions. - two new gates land as the proof the registry is real: ConformanceGate (frozen fixtures or no pass — previously only the publish guard checked this) and RegressionGate (flag-only compare against registry best-known; flags never block silently). - stage types: register_stage_type(); KNOWN_STAGES stays as a live alias, contracts.py never edits. - corpus source kinds: the Literal is gone (schema data); the planner's register_volume_provider decides what is mountable, and unknown kinds fail at plan time with the registered set. - license policy: LicensePolicy objects (allow_waivers) instead of a hardcoded predicate — policies vary by construction. 7 new tests written first and watched failing; suite 249 green, ruff clean. --- src/pipeline/contracts.py | 46 ++++++-- src/pipeline/gates.py | 176 +++++++++++++++++++++++++++---- src/pipeline/planner.py | 49 +++++++-- tests/test_pipeline_plan_gate.py | 10 +- tests/test_pipeline_registry.py | 159 ++++++++++++++++++++++++++++ 5 files changed, 404 insertions(+), 36 deletions(-) create mode 100644 tests/test_pipeline_registry.py diff --git a/src/pipeline/contracts.py b/src/pipeline/contracts.py index b244dc6e..7eae00c1 100644 --- a/src/pipeline/contracts.py +++ b/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: - 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]: diff --git a/src/pipeline/gates.py b/src/pipeline/gates.py index db97d0f6..ddd894e3 100644 --- a/src/pipeline/gates.py +++ b/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()] diff --git a/src/pipeline/planner.py b/src/pipeline/planner.py index 1ff166cb..9100f0f8 100644 --- a/src/pipeline/planner.py +++ b/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, diff --git a/tests/test_pipeline_plan_gate.py b/tests/test_pipeline_plan_gate.py index c2958d20..be581999 100644 --- a/tests/test_pipeline_plan_gate.py +++ b/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: diff --git a/tests/test_pipeline_registry.py b/tests/test_pipeline_registry.py new file mode 100644 index 00000000..08f2d170 --- /dev/null +++ b/tests/test_pipeline_registry.py @@ -0,0 +1,159 @@ +"""WO55 — extension-point registries: the spine's OCP round. + +Before this round, adding a gate, stage type, corpus source kind, or +license policy meant EDITING contracts/gates/planner. After it, each +is a registry: the modules are closed for modification, open for +extension — and the two new gates (conformance, regression) land as +the proof that extension needs no core edits. +""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +from pipeline.contracts import CorpusSpec, RecipeSpec, register_stage_type +from pipeline.gates import evaluate +from pipeline.planner import plan_stage + +REPO = Path(__file__).parent.parent +RECIPES = REPO / "recipes" +CORPORA = REPO / "corpora" + + +def _recipe_body() -> dict: + return { + "id": "xx-test", + "task": "transliteration", + "corpora": {"train": "xx-src@v1", "test": "xx-src@v1#test"}, + "model": "byt5-small", + "stages": [{"stage": "sft", "route": "legacy:train_xx.py", + "params": {"epochs": 1}}], + "gates": { + "quality": {"cer_max": 0.1}, + "convention": "row-mean", + "parity": {"agree_min": 0.98}, + }, + } + + +def test_register_stage_type_extends_contracts() -> None: + register_stage_type("my_custom_stage") + body = _recipe_body() + body["stages"] = [{"stage": "my_custom_stage", "params": {"x": 1}}] + r = RecipeSpec.model_validate(body) # no edit to contracts needed + assert r.stages[0].stage == "my_custom_stage" + + +def test_register_gate_extends_evaluate(tmp_path: Path) -> None: + from pipeline.gates import BaseGate, GateContext, GateResult, register_gate + + class AlwaysFlag(BaseGate): + name = "always_flag" + + def check(self, recipe: RecipeSpec, ctx: GateContext) -> GateResult: + return GateResult(self.name, True, "flagged by test") + + register_gate(AlwaysFlag()) + recipe = load_recipe(tmp_path) + results = evaluate(recipe, CORPORA, metrics={"cer": 0.05}, parity=None) + names = {g.name for g in results} + assert "always_flag" in names, "registry gate must appear without core edits" + assert "quality" in names and "parity" in names and "license" in names + + +def load_recipe(tmp_path: Path) -> RecipeSpec: + import yaml as _yaml + + (tmp_path / "recipes").mkdir(exist_ok=True) + (tmp_path / "recipes" / "xx-test.yaml").write_text( + _yaml.safe_dump(_recipe_body()), encoding="utf-8" + ) + from pipeline.contracts import load_recipe as _load + + return _load(tmp_path / "recipes" / "xx-test.yaml") + + +def test_conformance_gate_is_registered(tmp_path: Path) -> None: + recipe = load_recipe(tmp_path) + # missing conformance input -> gate fails loudly, not silently absent + results = evaluate(recipe, CORPORA, metrics={"cer": 0.05}, parity=None) + by_name = {g.name: g for g in results} + assert "conformance" in by_name + assert not by_name["conformance"].passed + # present and valid -> passes + import json + + vec = tmp_path / "vectors.json" + vec.write_text( + json.dumps({"model": "xx", "decode": {"mode": "single-row-greedy"}, + "vectors": [{"src": "a", "out": "b"}]}), + encoding="utf-8", + ) + ok = evaluate(recipe, CORPORA, metrics={"cer": 0.05}, parity=None, + conformance_path=vec) + assert [g for g in ok if g.name == "conformance"][0].passed + + +def test_regression_gate_flags_against_registry(tmp_path: Path) -> None: + recipe = load_recipe(tmp_path) + registry = {"xx-test": {"best": {"cer": 0.01}}} + results = evaluate(recipe, CORPORA, metrics={"cer": 0.05}, parity=None, + registry=registry) + reg = [g for g in results if g.name == "regression"][0] + assert reg.passed, "regression is flag-only (owner waives), never blocks" + assert "0.05" in reg.detail and "0.01" in reg.detail + # better-than-registry -> clean detail + better = evaluate(recipe, CORPORA, metrics={"cer": 0.005}, parity=None, + registry=registry) + assert "improves" in [g for g in better if g.name == "regression"][0].detail + + +def test_volume_provider_registry_extends_planner(tmp_path: Path) -> None: + from pipeline.planner import register_volume_provider + from pipeline.runner import Volume + + register_volume_provider("test-kind", + lambda src, idx: Volume(f"test://{src.id}", + f"/test{idx}", "ro")) + corpus = { + "id": "xx-src", + "task": "transliteration", + "sources": [{"kind": "test-kind", "id": "org/ds", "license": "CC0-1.0"}], + "splits": { + "train": {"path": "train.jsonl", "rows": 10}, + "test": {"path": "test.jsonl", "rows": 5}, + }, + } + CorpusSpec.model_validate(corpus) # kinds are schema data, not a Literal + recipe_body = _recipe_body() + (tmp_path / "corpora").mkdir(exist_ok=True) + (tmp_path / "corpora" / "xx-src.yaml").write_text(yaml.safe_dump(corpus)) + (tmp_path / "recipes").mkdir(exist_ok=True) + (tmp_path / "recipes" / "xx-test.yaml").write_text(yaml.safe_dump(recipe_body)) + recipe = load_recipe(tmp_path) # module helper: writes + loads the recipe + spec = plan_stage(recipe, recipe.stages[0], run="r1", + corpora_dir=tmp_path / "corpora") + urls = {v.url for v in spec.volumes} + assert "test://org/ds" in urls, "provider registry must mount new kinds" + + +def test_license_policy_is_encapsulated() -> None: + from pipeline.contracts import CorpusSource + + src = CorpusSource.model_validate({ + "kind": "hf-dataset", "id": "o/d", "license": "LicenseRef-x", + "license_review": {"status": "owner-approved", + "date": "2026-10-19", "note": "n"}, + }) + from pipeline.contracts import DEFAULT_LICENSE_POLICY + + assert not DEFAULT_LICENSE_POLICY.blocked(src) + restricted = CorpusSource.model_validate({ + "kind": "hf-dataset", "id": "o/d", "license": "LicenseRef-y", + "license_review": {"status": "owner-approved", + "date": "2026-10-19", "note": "n"}, + }) + strict = type(DEFAULT_LICENSE_POLICY)(allow_waivers=False) + assert strict.blocked(restricted), "policies vary without core edits"