Feat/koth engine lane

.github/scripts/koth_score.py

@@ -0,0 +1,108 @@
+"""Paired champion-vs-challenger scoring for the koth ladder.
+
+Runs vouchbench twice per seed — once with the reigning kit, once with the
+challenger kit — over seeds derived from the base sha and the utc date, and
+applies the dethrone test from docs/vouchbench-seasons.md:
+
+    dethroned  iff  mean(challenger - champion)  >=  max(0.007, 1.96 x SE)
+
+where SE is the standard error of the per-seed paired differences (common
+random numbers: both arms see identical generated sessions per seed, so the
+difference cancels seed-to-seed variance).
+
+Seeds are deterministic for a given (base sha, utc day): anyone can recompute
+the day's seeds and reproduce the run locally. Same-day overfitting to known
+seeds is bounded by the margin band and settled monthly by the season's
+commit-reveal scored run, which uses seeds that do not exist until the
+cutoff.
+
+Usage:
+    python koth_score.py --champion kit.yaml --challenger kit-pr.yaml \
+        --base-sha <sha> [--date YYYY-MM-DD] [--out report.json]
+
+Exit code 0 = dethroned (challenger wins), 3 = held (champion stands),
+1 = error. The distinct win/hold codes let the workflow branch without
+parsing the report.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import hashlib
+import json
+from pathlib import Path
+
+from vouch import bench
+
+# 12 paired seeds, not 4: the band is built from the standard error of the
+# paired differences, and n=4 makes that SE noisy and its normal-z reading
+# optimistic. z=1.96 is the two-sided 95% normal quantile; with a small n
+# the paired diffs are not perfectly normal, so the floor below still does
+# the real gatekeeping when the SE collapses on a deterministic overfit.
+N_SEEDS = 12
+FLOOR = bench.DETHRONE_FLOOR
+Z = bench.DETHRONE_Z
+# bench sleeps this long between session ingests to spread created_at
+# timestamps; keep it small — it is wall-clock, not simulated time. the
+# 3600s that lived here briefly would have out-slept the CI job timeout
+# many times over.
+SESSION_GAP_SECONDS = 2.0
+
+
+def day_seeds(base_sha: str, date: str, n: int = N_SEEDS) -> list[int]:
+    """Derive the day's seed list from the champion sha and the utc date."""
+    seeds = []
+    for i in range(n):
+        digest = hashlib.sha256(f"{base_sha}:{date}:{i}".encode()).hexdigest()
+        seeds.append(int(digest[:12], 16))
+    return seeds
+
+
+def score(kit_text: str, seeds: list[int]) -> list[float]:
+    extra = kit_text if kit_text.strip() else None
+    return [
+        bench.run(
+            s, extra_config=extra, session_gap_seconds=SESSION_GAP_SECONDS
+        )["composite"]
+        for s in seeds
+    ]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--champion", required=True)
+    parser.add_argument("--challenger", required=True)
+    parser.add_argument("--base-sha", required=True)
+    parser.add_argument("--date", default=None)
+    parser.add_argument("--out", default=None)
+    args = parser.parse_args()
+
+    date = args.date or dt.datetime.now(dt.UTC).strftime("%Y-%m-%d")
+    seeds = day_seeds(args.base_sha, date)
+
+    champion_text = Path(args.champion).read_text(encoding="utf-8")
+    challenger_text = Path(args.challenger).read_text(encoding="utf-8")
+
+    champion_scores = score(champion_text, seeds)
+    challenger_scores = score(challenger_text, seeds)
+    verdict = bench.paired_verdict(
+        champion_scores, challenger_scores, floor=FLOOR, z=Z,
+    )
+
+    report = {
+        "date": date,
+        "base_sha": args.base_sha,
+        "seeds": seeds,
+        "lane": "kit",
+        **verdict,
+    }
+    text = json.dumps(report, indent=1)
+    print(text)
+    if args.out:
+        Path(args.out).write_text(text + "\n", encoding="utf-8")
+    return 0 if report["dethroned"] else 3
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/scripts/score_strategy.py

@@ -0,0 +1,82 @@
+"""Paired champion-vs-challenger scoring for the engine (strategy) lane.
+
+Same dethrone test as the kit lane (docs/vouchbench-seasons.md), but the arm
+is a pluggable ranking strategy instead of a config fragment: each submission
+is loaded as an untrusted file and run through vouch.strategy.SandboxProxy, so
+the scored code executes only inside the sandbox child.
+
+    dethroned  iff  mean(challenger - champion)  >=  max(0.007, 1.96 x SE)
+
+Exit code 0 = dethroned, 3 = held, 1 = error. Unlike the kit lane, a winning
+verdict here does NOT auto-merge: engine code ships only through human review.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import hashlib
+import json
+from pathlib import Path
+
+from vouch import bench
+from vouch.strategy import SandboxProxy
+
+N_SEEDS = 8
+FLOOR = bench.DETHRONE_FLOOR
+Z = bench.DETHRONE_Z
+SESSION_GAP_SECONDS = 2.0
+
+
+def day_seeds(base_sha: str, date: str, n: int = N_SEEDS) -> list[int]:
+    seeds = []
+    for i in range(n):
+        digest = hashlib.sha256(f"{base_sha}:{date}:{i}".encode()).hexdigest()
+        seeds.append(int(digest[:12], 16))
+    return seeds
+
+
+def score(path: str, seeds: list[int]) -> list[float]:
+    proxy = SandboxProxy(path)
+    return [
+        bench.run(s, strategy=proxy, session_gap_seconds=SESSION_GAP_SECONDS)[
+            "composite"
+        ]
+        for s in seeds
+    ]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--champion", required=True)
+    parser.add_argument("--challenger", required=True)
+    parser.add_argument("--base-sha", required=True)
+    parser.add_argument("--date", default=None)
+    parser.add_argument("--out", default=None)
+    args = parser.parse_args()
+
+    date = args.date or dt.datetime.now(dt.UTC).strftime("%Y-%m-%d")
+    seeds = day_seeds(args.base_sha, date)
+
+    champion_scores = score(args.champion, seeds)
+    challenger_scores = score(args.challenger, seeds)
+    verdict = bench.paired_verdict(
+        champion_scores, challenger_scores, floor=FLOOR, z=Z,
+    )
+
+    report = {
+        "date": date,
+        "base_sha": args.base_sha,
+        "seeds": seeds,
+        "lane": "engine",
+        **verdict,
+    }
+    text = json.dumps(report, indent=1)
+    print(text)
+    if args.out:
+        Path(args.out).write_text(text + "\n", encoding="utf-8")
+    return 0 if report["dethroned"] else 3
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/scripts/update_leaderboard.py

@@ -0,0 +1,84 @@
+"""Append a dethrone row to competition/LEADERBOARD.md from a scorecard.
+
+The ledger row was appended by hand at season close; every dethrone that
+lands on the ladder branch now writes its own row. Input is the report
+json both scorers emit (koth_score.py for kits, score_strategy.py for
+strategies — same shape, ``lane`` distinguishes them).
+
+Idempotent on the PR number: if the ledger already cites the PR, the
+script exits 0 without writing, so the ledger workflow re-running on its
+own push converges instead of looping.
+
+Usage:
+    python update_leaderboard.py --report report.json --pr 123 \
+        --author somebody [--ledger competition/LEADERBOARD.md]
+
+Exit code 0 = row appended or already present, 1 = error.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from pathlib import Path
+
+
+def next_row_number(ledger_text: str) -> int:
+    rows = re.findall(r"^\| (\d+) \|", ledger_text, flags=re.M)
+    return max((int(r) for r in rows), default=-1) + 1
+
+
+def format_row(
+    number: int, report: dict, pr: int, author: str
+) -> str:
+    lane = report.get("lane", "kit")
+    date = report.get("date", "")
+    mean = report["challenger"]["mean"]
+    margin = report["mean_diff"]
+    return (
+        f"| {number} | {author} ({lane}) | #{pr} | {date} "
+        f"| {mean:.4f} | +{margin:.4f} |"
+    )
+
+
+def append_row(ledger: Path, report: dict, pr: int, author: str) -> bool:
+    """Add the row unless the PR is already cited. True = file changed."""
+    text = ledger.read_text(encoding="utf-8")
+    if re.search(rf"\| #{pr} \|", text):
+        return False
+    row = format_row(next_row_number(text), report, pr, author)
+    if not text.endswith("\n"):
+        text += "\n"
+    # the table is the last block before the payout footer; append right
+    # after the final table row so the footer prose stays at the bottom
+    lines = text.splitlines(keepends=True)
+    last_row = max(
+        i for i, line in enumerate(lines) if line.startswith("| ")
+    )
+    lines.insert(last_row + 1, row + "\n")
+    ledger.write_text("".join(lines), encoding="utf-8")
+    return True
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--report", required=True)
+    parser.add_argument("--pr", required=True, type=int)
+    parser.add_argument("--author", required=True)
+    parser.add_argument(
+        "--ledger", default="competition/LEADERBOARD.md",
+    )
+    args = parser.parse_args()
+
+    report = json.loads(Path(args.report).read_text(encoding="utf-8"))
+    if not report.get("dethroned"):
+        print("report is not a dethrone; nothing to append")
+        return 0
+    changed = append_row(Path(args.ledger), report, args.pr, args.author)
+    print("row appended" if changed else f"pr #{args.pr} already in ledger")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/scripts/validate_kit.py

@@ -0,0 +1,107 @@
+"""Validate a ladder kit file against the strict allowlist.
+
+The kit is the one file an untrusted PR may change, and it is applied as
+extra_config inside a write-token CI context — so the surface must be data
+only. In particular, config keys that name executables (compile.llm_cmd,
+enrich.llm_cmd) must never be reachable from a kit: a kit that could set a
+command string would execute it on the runner. The allowlist below is
+therefore closed-world: unknown keys are an error, not a warning.
+
+Usage: python validate_kit.py <kit.yaml>
+Exits 0 and prints the canonical yaml on success; exits 1 with a reason
+on any violation.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+MAX_KIT_BYTES = 4096
+
+BACKENDS = {"auto", "fts5", "embedding", "hybrid", "substring"}
+
+# key path -> (type, validator)
+BOUNDS: dict[tuple[str, ...], Any] = {
+    ("retrieval", "backend"): lambda v: isinstance(v, str) and v in BACKENDS,
+    ("retrieval", "default_limit"): lambda v: (
+        isinstance(v, int) and not isinstance(v, bool) and 1 <= v <= 100
+    ),
+    ("retrieval", "prompt_gate", "enabled"): lambda v: isinstance(v, bool),
+    ("retrieval", "recency", "enabled"): lambda v: isinstance(v, bool),
+    ("retrieval", "recency", "half_life_days"): lambda v: (
+        isinstance(v, (int, float))
+        and not isinstance(v, bool)
+        and 0.01 <= float(v) <= 3650.0
+    ),
+    ("retrieval", "rerank", "enabled"): lambda v: isinstance(v, bool),
+    ("retrieval", "rerank", "top_k"): lambda v: (
+        isinstance(v, int) and not isinstance(v, bool) and 1 <= v <= 500
+    ),
+    ("retrieval", "pages_first", "enabled"): lambda v: isinstance(v, bool),
+    ("retrieval", "pages_first", "boost"): lambda v: (
+        isinstance(v, (int, float))
+        and not isinstance(v, bool)
+        and 0.1 <= float(v) <= 10.0
+    ),
+}
+
+ALLOWED_BRANCHES = {path[:i] for path in BOUNDS for i in range(1, len(path))}
+
+
+def _walk(node: Any, path: tuple[str, ...], errors: list[str]) -> None:
+    if isinstance(node, dict):
+        for key, value in node.items():
+            if not isinstance(key, str):
+                errors.append(f"non-string key at {'.'.join(path) or '<root>'}")
+                continue
+            child = (*path, key)
+            if child in BOUNDS:
+                if not BOUNDS[child](value):
+                    errors.append(f"{'.'.join(child)}: value {value!r} out of bounds")
+            elif child in ALLOWED_BRANCHES:
+                _walk(value, child, errors)
+            else:
+                errors.append(f"{'.'.join(child)}: key not in allowlist")
+    else:
+        errors.append(f"{'.'.join(path) or '<root>'}: expected a mapping")
+
+
+def validate(text: str) -> list[str]:
+    # an empty kit must NOT validate: over the contents-api, a file larger
+    # than the api's inline limit comes back with empty content, which would
+    # otherwise decode to "" and pass vacuously — the PR would then be scored
+    # as engine defaults rather than its real, oversized contents. require a
+    # real retrieval mapping so "empty" can never mean "champion defaults".
+    encoded = text.encode("utf-8")
+    if not encoded.strip():
+        return ["empty kit (nothing to score — did the fetch truncate?)"]
+    if len(encoded) > MAX_KIT_BYTES:
+        return [f"kit larger than {MAX_KIT_BYTES} bytes"]
+    try:
+        data = yaml.safe_load(text)
+    except yaml.YAMLError as exc:
+        return [f"not valid yaml: {exc}"]
+    if not isinstance(data, dict) or "retrieval" not in data:
+        return ["kit must be a mapping with a top-level 'retrieval' section"]
+    errors: list[str] = []
+    _walk(data, (), errors)
+    return errors
+
+
+def main() -> int:
+    text = Path(sys.argv[1]).read_text(encoding="utf-8")
+    errors = validate(text)
+    if errors:
+        for err in errors:
+            print(f"kit rejected: {err}", file=sys.stderr)
+        return 1
+    print(text)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())