feat(priors): cold-start σ_raw prior table — 78 cited per-model cells

scripts/parity_runner.js

@@ -9,6 +9,7 @@
 const fs = require("fs");
 const MSO = require("../web/mso-core.js");
 const MSOEstimate = require("../web/mso-estimate.js");
+const MSO_Priors = require("../web/mso-priors.js");
 
 const inPath = process.argv[2];
 const outPath = process.argv[3];
@@ -59,4 +60,18 @@ out.risk = cases.risk.map((p) => MSO.rankNodesByRisk(p).map((r) => ({
   fi: r.fan_in_degree, fo: r.fan_out_degree, bott: r.is_bottleneck,
 })));
 
+out.seed_node = (cases.seed_node || []).map((c) => {
+  const s = MSO_Priors.seedNode(c.model, c.task_type);
+  const p = s.provenance;
+  return {
+    seeds: s.seeds,
+    sigma_skill: s.sigma_skill != null ? s.sigma_skill : null,
+    catch_rate: s.catch_rate,
+    confidence: p.confidence,
+    band_low: p.band.low,
+    band_mid: p.band.mid,
+    band_high: p.band.high,
+  };
+});
+
 fs.writeFileSync(outPath, JSON.stringify(out));

src/minimal_oversight/priors.py

@@ -0,0 +1,181 @@
+"""Cold-start task-competence priors for MSO.
+
+When a practitioner has no traces yet, they still need *starting* numbers to seed
+a pipeline node. This module loads a curated, provenance-bound prior table
+(`data/priors.yaml`) and maps a ``(model, task_type)`` cell onto the quantities a
+:class:`~minimal_oversight.models.Node` needs:
+
+- generator task-types -> ``sigma_skill`` (via ``sigma_raw_mid / gamma``), so the
+  model reproduces the prior's ``sigma_raw`` at the return-operator fixed point;
+- the ``review`` task-type -> ``catch_rate`` (how well the model catches an
+  upstream node's errors).
+
+Priors are *hypotheses*, not ground truth. Each carries a low/mid/high band and a
+source; a wide band means weak evidence. They are meant to be refined by the
+user's own logged outcomes — see :mod:`minimal_oversight.estimation`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from functools import lru_cache
+from importlib import resources
+from typing import Any
+
+import yaml
+
+REVIEW_TASK = "review"
+
+
+@dataclass(frozen=True)
+class Band:
+    """A low/mid/high estimate. ``width`` is a crude evidence-strength proxy."""
+
+    low: float
+    mid: float
+    high: float
+
+    @property
+    def width(self) -> float:
+        return self.high - self.low
+
+    @classmethod
+    def from_mapping(cls, m: dict[str, float]) -> Band:
+        return cls(low=float(m["low"]), mid=float(m["mid"]), high=float(m["high"]))
+
+
+@dataclass(frozen=True)
+class PriorCell:
+    """One ``(model, task_type)`` prior with provenance."""
+
+    model: str
+    task_type: str
+    sigma_raw: Band | None          # generator task-types
+    catch_rate: Band | None         # the `review` task-type
+    primary_benchmark: str | None
+    metric: str | None
+    metric_kind: str | None         # "absolute" | "relative"
+    reported_value: Any
+    sample_size: Any
+    observed_date: str | None
+    source_url: str | None
+    normalization_note: str | None
+
+
+def _clamp(x: float, lo: float, hi: float) -> float:
+    return max(lo, min(hi, x))
+
+
+@lru_cache(maxsize=4)
+def load_priors(path: str | None = None) -> dict[str, Any]:
+    """Load and lightly validate the prior table.
+
+    Returns a dict with ``meta``, ``task_types`` and a ``cells`` map keyed by
+    ``(model, task_type)``. Result is cached; pass an explicit ``path`` to bypass
+    the packaged file (e.g. in tests).
+    """
+    if path is None:
+        text = resources.files("minimal_oversight.data").joinpath("priors.yaml").read_text()
+    else:
+        with open(path, encoding="utf-8") as fh:
+            text = fh.read()
+    raw = yaml.safe_load(text)
+
+    meta = raw.get("meta", {})
+    task_types = list(raw.get("task_types", []))
+    gamma = float(meta.get("gamma", 10 / 12))
+    if not (0.0 < gamma <= 1.0):
+        raise ValueError(f"priors meta.gamma out of range: {gamma}")
+
+    cells: dict[tuple[str, str], PriorCell] = {}
+    for row in raw.get("priors", []):
+        model = str(row["model"])
+        task = str(row["task_type"])
+        if task not in task_types:
+            raise ValueError(f"unknown task_type {task!r} for model {model!r}")
+        sigma = Band.from_mapping(row["sigma_raw"]) if "sigma_raw" in row else None
+        catch = Band.from_mapping(row["catch_rate"]) if "catch_rate" in row else None
+        if task == REVIEW_TASK and catch is None:
+            raise ValueError(f"review cell {model!r} must carry a catch_rate band")
+        if task != REVIEW_TASK and sigma is None:
+            raise ValueError(f"generator cell {model!r}/{task!r} must carry a sigma_raw band")
+        for b in (sigma, catch):
+            if b is not None and not (0.0 <= b.low <= b.mid <= b.high <= 1.0):
+                raise ValueError(f"non-monotone or out-of-range band for {model!r}/{task!r}")
+        key = (model, task)
+        if key in cells:
+            raise ValueError(f"duplicate prior cell {key}")
+        cells[key] = PriorCell(
+            model=model,
+            task_type=task,
+            sigma_raw=sigma,
+            catch_rate=catch,
+            primary_benchmark=row.get("primary_benchmark"),
+            metric=row.get("metric"),
+            metric_kind=row.get("metric_kind"),
+            reported_value=row.get("reported_value"),
+            sample_size=row.get("sample_size"),
+            observed_date=row.get("observed_date"),
+            source_url=row.get("source_url"),
+            normalization_note=row.get("normalization_note"),
+        )
+    return {"meta": meta, "gamma": gamma, "task_types": task_types, "cells": cells}
+
+
+def list_models(path: str | None = None) -> list[str]:
+    """Sorted distinct model names in the table."""
+    return sorted({m for (m, _t) in load_priors(path)["cells"]})
+
+
+def list_task_types(path: str | None = None) -> list[str]:
+    return list(load_priors(path)["task_types"])
+
+
+def get_cell(model: str, task_type: str, path: str | None = None) -> PriorCell:
+    cells = load_priors(path)["cells"]
+    try:
+        return cells[(model, task_type)]
+    except KeyError:
+        raise KeyError(f"no prior for model={model!r} task_type={task_type!r}") from None
+
+
+def seed_node(model: str, task_type: str, path: str | None = None) -> dict[str, Any]:
+    """Map a prior cell onto ``Node`` keyword arguments + provenance.
+
+    For generator task-types returns ``sigma_skill`` (so the model reproduces the
+    prior's ``sigma_raw`` at the fixed point) and a zero ``catch_rate``. For the
+    ``review`` task-type returns the ``catch_rate`` (and no ``sigma_skill``; the
+    caller pairs the reviewer with the node it corrects).
+
+    The returned ``provenance`` block carries the band, source, and a
+    ``confidence`` proxy (``1 - band_width``) so the UI can render priors as
+    hypotheses, never as ground truth.
+    """
+    data = load_priors(path)
+    gamma = data["gamma"]
+    cell = get_cell(model, task_type, path)
+
+    out: dict[str, Any] = {"model": model, "task_type": task_type, "is_prior": True}
+    if task_type == REVIEW_TASK:
+        band = cell.catch_rate
+        out["catch_rate"] = _clamp(band.mid, 0.0, 1.0)
+        out["seeds"] = "catch_rate"
+    else:
+        band = cell.sigma_raw
+        meta = data["meta"]
+        default_catch = float(meta.get("generator_default_catch_rate", 0.0))
+        out["sigma_skill"] = _clamp(band.mid / gamma, 0.05, 0.98)
+        out["catch_rate"] = default_catch
+        out["seeds"] = "sigma_skill"
+
+    out["provenance"] = {
+        "band": {"low": band.low, "mid": band.mid, "high": band.high},
+        "confidence": round(1.0 - band.width, 3),
+        "primary_benchmark": cell.primary_benchmark,
+        "metric": cell.metric,
+        "metric_kind": cell.metric_kind,
+        "observed_date": cell.observed_date,
+        "source_url": cell.source_url,
+        "normalization_note": cell.normalization_note,
+    }
+    return out

tests/test_parity.py

@@ -20,6 +20,7 @@
 from minimal_oversight import _formulae as F
 from minimal_oversight import analyze_pipeline
 from minimal_oversight import estimation as E
+from minimal_oversight import priors as P
 from minimal_oversight.allocation import select_scope
 from minimal_oversight.capacity import check_feasibility
 from minimal_oversight.models import AggregationType, Node, PipelineGraph
@@ -122,6 +123,14 @@ def _nd(node_id, skill, catch, parents, agg="product"):
         {"sigma": [0.667, 0.517, 0.417, 0.375, 0.708, 0.542], "p_min": 0.5, "coverage": 0.0},
         {"sigma": [0.8, 0.4, 0.6, 0.3, 0.7], "p_min": 0.6, "coverage": 0.5},
     ],
+    # mso-priors.js seeds a node from (model × task) benchmark priors.
+    # Covers generator branch (sigma_skill) and review branch (catch_rate).
+    "seed_node": [
+        {"model": "claude-opus-4", "task_type": "code_generation"},
+        {"model": "gpt-4o", "task_type": "drafting"},
+        {"model": "gemini-2-flash", "task_type": "review"},
+        {"model": "deepseek-r1", "task_type": "extraction"},
+    ],
     # web/mso-estimate.js turns a practitioner's real outcomes into per-node
     # sigma_raw / sigma_corr / catch / masking. Pin those to estimation.py.
     "estimate": [
@@ -250,6 +259,19 @@ def _python_expected() -> dict:
             }
             for r in rep.node_risks
         ])
+    exp["seed_node"] = []
+    for c in CASES["seed_node"]:
+        s = P.seed_node(c["model"], c["task_type"])
+        prov = s["provenance"]
+        exp["seed_node"].append({
+            "seeds": s["seeds"],
+            "sigma_skill": s.get("sigma_skill"),
+            "catch_rate": s["catch_rate"],
+            "confidence": prov["confidence"],
+            "band_low": prov["band"]["low"],
+            "band_mid": prov["band"]["mid"],
+            "band_high": prov["band"]["high"],
+        })
     return exp
 
 
@@ -318,3 +340,13 @@ def test_browser_port_matches_python_reference():
             assert gn["fi"] == en["fi"] and gn["fo"] == en["fo"] and gn["bott"] == en["bott"]
             for k in ["sota", "dc", "masking"]:
                 assert _close(gn[k], en[k]), f"risk {en['name']} {k} mismatch"
+
+    for i, (g, e) in enumerate(zip(got["seed_node"], exp["seed_node"])):
+        c = CASES["seed_node"][i]
+        label = f"{c['model']}|{c['task_type']}"
+        assert g["seeds"] == e["seeds"], f"seed_node seeds mismatch for {label}"
+        for k in ["band_low", "band_mid", "band_high", "confidence"]:
+            assert _close(g[k], e[k]), f"seed_node {k} mismatch for {label}"
+        if e["sigma_skill"] is not None:
+            assert _close(g["sigma_skill"], e["sigma_skill"]), f"seed_node sigma_skill mismatch for {label}"
+        assert _close(g["catch_rate"], e["catch_rate"]), f"seed_node catch_rate mismatch for {label}"

tests/test_priors.py

@@ -0,0 +1,103 @@
+"""Tests for the cold-start priors loader and the (model, task) -> Node seeding."""
+
+from __future__ import annotations
+
+import textwrap
+
+import pytest
+
+from minimal_oversight import priors as P
+
+
+def test_table_loads_and_lists():
+    models = P.list_models()
+    tasks = P.list_task_types()
+    assert models, "expected at least one model in the prior table"
+    assert "review" in tasks
+    assert "code_generation" in tasks
+
+
+def test_generator_seed_reproduces_sigma_raw_at_fixed_point():
+    """sigma_skill is set so gamma * sigma_skill == the prior's sigma_raw mid."""
+    data = P.load_priors()
+    gamma = data["gamma"]
+    for (model, task), cell in data["cells"].items():
+        if task == "review":
+            continue
+        seed = P.seed_node(model, task)
+        assert seed["seeds"] == "sigma_skill"
+        assert 0.05 <= seed["sigma_skill"] <= 0.98
+        assert seed["catch_rate"] == 0.0  # a generator does not correct its parent
+        # round-trips unless the mid was clamped at the [0.05, 0.98] rails
+        unclamped = cell.sigma_raw.mid / gamma
+        if 0.05 <= unclamped <= 0.98:
+            assert gamma * seed["sigma_skill"] == pytest.approx(cell.sigma_raw.mid, abs=1e-9)
+
+
+def test_review_seed_yields_catch_rate():
+    review_cells = [(m, t) for (m, t) in P.load_priors()["cells"] if t == "review"]
+    assert review_cells, "expected at least one review cell"
+    for model, _ in review_cells:
+        seed = P.seed_node(model, "review")
+        assert seed["seeds"] == "catch_rate"
+        assert 0.0 <= seed["catch_rate"] <= 1.0
+        assert "sigma_skill" not in seed
+
+
+def test_provenance_block_is_well_formed():
+    for (model, task) in P.load_priors()["cells"]:
+        prov = P.seed_node(model, task)["provenance"]
+        assert 0.0 <= prov["confidence"] <= 1.0
+        b = prov["band"]
+        assert 0.0 <= b["low"] <= b["mid"] <= b["high"] <= 1.0
+        assert prov["metric_kind"] in {"absolute", "relative", None}
+
+
+def test_unknown_cell_raises():
+    with pytest.raises(KeyError):
+        P.seed_node("no_such_model", "code_generation")
+
+
+def _write(tmp_path, body: str) -> str:
+    p = tmp_path / "priors.yaml"
+    p.write_text(textwrap.dedent(body))
+    return str(p)
+
+
+def test_loader_rejects_nonmonotone_band(tmp_path):
+    path = _write(tmp_path, """
+        meta: {schema_version: 1, gamma: 0.8333333333333334, status: test}
+        task_types: [code_generation, review]
+        priors:
+          - model: m
+            task_type: code_generation
+            sigma_raw: {low: 0.8, mid: 0.5, high: 0.9}
+    """)
+    with pytest.raises(ValueError):
+        P.load_priors(path)
+
+
+def test_loader_rejects_review_without_catch_rate(tmp_path):
+    path = _write(tmp_path, """
+        meta: {schema_version: 1, gamma: 0.8333333333333334, status: test}
+        task_types: [code_generation, review]
+        priors:
+          - model: m
+            task_type: review
+            sigma_raw: {low: 0.4, mid: 0.6, high: 0.8}
+    """)
+    with pytest.raises(ValueError):
+        P.load_priors(path)
+
+
+def test_loader_rejects_unknown_task_type(tmp_path):
+    path = _write(tmp_path, """
+        meta: {schema_version: 1, gamma: 0.8333333333333334, status: test}
+        task_types: [code_generation]
+        priors:
+          - model: m
+            task_type: telepathy
+            sigma_raw: {low: 0.4, mid: 0.6, high: 0.8}
+    """)
+    with pytest.raises(ValueError):
+        P.load_priors(path)