diff --git a/tests/test_tuning_tools.py b/tests/test_tuning_tools.py index e232f833..dd7b1ddb 100644 --- a/tests/test_tuning_tools.py +++ b/tests/test_tuning_tools.py @@ -35,6 +35,7 @@ def load(name, relative): ctt = load("ctt_fastchess_shim", "tools/tune/chess_tuning_tools/fastchess_shim.py") ctt_config = load("ctt_make_config", "tools/tune/chess_tuning_tools/make_config.py") +audit_consensus = load("audit_consensus", "tools/tune/audit_consensus.py") clop = load("clop_fastchess", "tools/tune/clop/clop_fastchess.py") calibrate_panel = load("calibrate_panel", "tools/tune/calibrate_panel.py") gating = load("gating", "tools/tune/gating.py") @@ -62,6 +63,50 @@ def load(name, relative): class TuningToolsTest(unittest.TestCase): + def test_consensus_audit_recomputes_selection(self): + parameters = { + "X": {"name": "X", "type": "discrete", "default": 0, "values": [0, 1, 2]}, + "Y": {"name": "Y", "type": "discrete", "default": 0, "values": [0, 1]}, + } + space = {"parameters": list(parameters.values()), "conditions": []} + a, b, c = {"X": 0, "Y": 0}, {"X": 2, "Y": 0}, {"X": 2, "Y": 1} + payload = json.dumps(a, sort_keys=True, separators=(",", ":")) + candidate = { + "lane_optima": [a] * 10 + [b] * 5 + [c] * 5, + "selected": a, + "selection_evidence": { + "support": 10, + "total_normalized_l1": 15.0, + "canonical_sha256": hashlib.sha256(payload.encode()).hexdigest(), + }, + } + audit_consensus.verify_selection(candidate, parameters, space) + candidate["selected"] = b + with self.assertRaisesRegex(RuntimeError, "not the preregistered consensus"): + audit_consensus.verify_selection(candidate, parameters, space) + + def test_consensus_audit_null_limit_disables_both_null_modes(self): + selected = { + "QS": 40, "NULL_LIMIT": 0, "NULL_SPAN": 3, "NULL_MIN_DEPTH": 2, + "FUEL_NULL": 1, "FUEL_MIN_DEPTH": 6, "LMR_LIMIT": 750, + "LMR_MIN_DEPTH": 6, "FUT_CAP_DEPTH": 3, "IID": 0, + } + status = audit_consensus.mechanism_status(selected) + self.assertTrue(status["scoring_null_off"]) + self.assertTrue(status["fuel_null_off"]) + self.assertTrue(status["all_null_off"]) + + def test_consensus_compact_policy_drops_line_costing_iid(self): + selected = {"IID": 1, "IID_MIN_DEPTH": 8, "IID_RED": 5} + parameters = { + "IID": {"default": 0}, + "IID_MIN_DEPTH": {"default": 3}, + "IID_RED": {"default": 3}, + } + self.assertEqual( + audit_consensus.compact_policy(selected, parameters), + {"IID": 0, "IID_MIN_DEPTH": 3, "IID_RED": 3}) + def test_frozen_recovery_manifest_is_self_consistent(self): manifest, space = verify_recovery.audit(root=ROOT) self.assertEqual(manifest["budget"]["games"], 1000) diff --git a/tools/tune/README.md b/tools/tune/README.md index 79591a55..409b1e79 100644 --- a/tools/tune/README.md +++ b/tools/tune/README.md @@ -29,6 +29,8 @@ categorical, and Boolean parameters. budgets. It can also pool pairwise-only SPSA studies. - `freeze_recommendations.py` normalizes and audits a complete method/start/ checkpoint recommendation grid before held-out games are consulted. +- `audit_consensus.py` independently recomputes the frozen support/L1/SHA + consensus and its eventual-correctness parameters before validation. - `validate.py` measures recommendations on an independent opening set. - `plot_recovery.py` produces held-out Elo-versus-training-games curves and a paired method-comparison CSV at the primary checkpoint. diff --git a/tools/tune/audit_consensus.py b/tools/tune/audit_consensus.py new file mode 100644 index 00000000..22930831 --- /dev/null +++ b/tools/tune/audit_consensus.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +"""Audit a sealed Sunfish tuning consensus without consulting validation games.""" + +import argparse +import hashlib +import json +import pathlib + + +def digest(path): + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def verify_seal(root): + manifest = root / "manifest-sha256.txt" + seal = root / "SEALED" + if not manifest.is_file() or not seal.is_file(): + raise RuntimeError(f"unsealed artifact: {root}") + fields = dict(line.split(maxsplit=1) for line in seal.read_text().splitlines()[1:]) + if fields.get("manifest-sha256.txt") != digest(manifest): + raise RuntimeError("seal does not bind manifest") + for line in manifest.read_text().splitlines(): + expected, relative = line.split(" ", 1) + path = root / relative + if not path.is_file() or digest(path) != expected: + raise RuntimeError(f"manifest mismatch: {relative}") + + +def domain(record): + values = record.get("ordered_values", record.get("values")) + if values is None and record["type"] == "integer": + values = range(record["min"], record["max"] + 1, record.get("step", 1)) + if values is None: + raise RuntimeError(f"unsupported parameter domain: {record['name']}") + return list(values) + + +def canonical(options, space): + defaults = {record["name"]: record["default"] for record in space["parameters"]} + result = dict(options) + for _ in range(len(space.get("conditions", [])) + 1): + old = dict(result) + for clause in space.get("conditions", []): + if all(result[name] in values for name, values in clause["when"].items()): + for name in clause.get("reset", []): + result[name] = defaults[name] + result.update(clause.get("set", {})) + if result == old: + return result + raise RuntimeError("space canonicalization did not converge") + + +def verify_eventual_space(parameters): + domains = {name: domain(record) for name, record in parameters.items()} + positive_reductions = ("NULL_CUT_RED", "NULL_RED", "IID_RED") + if any(value < 1 for name in positive_reductions for value in domains[name]): + raise RuntimeError("recursive probe reductions must remain positive") + if any(value < 0 for value in domains["NULL_SPAN"]): + raise RuntimeError("the scoring-null interval must remain finite") + if any(value not in (0, 1, 2) for value in domains["FUEL_NULL"]): + raise RuntimeError("fuel debt lies outside the proved bounded domain") + if any(value not in (1, 2) for value in domains["LMR_RED"]): + raise RuntimeError("LMR debt lies outside the proved bounded domain") + + +def verify_selection(candidate, parameters, space): + names = list(parameters) + domains = {name: domain(record) for name, record in parameters.items()} + lanes = [canonical(options, space) for options in candidate["lane_optima"]] + if len(lanes) != 20: + raise RuntimeError("combined consensus must contain twenty lane optima") + + def coordinate(options): + return tuple(domains[name].index(options[name]) / max(1, len(domains[name]) - 1) + for name in names) + + unique = {} + for options in lanes: + payload = json.dumps(options, sort_keys=True, separators=(",", ":")) + unique.setdefault(payload, {"options": options, "support": 0})["support"] += 1 + points = [coordinate(options) for options in lanes] + choices = [] + for payload, record in unique.items(): + point = coordinate(record["options"]) + distance = sum(sum(abs(a - b) for a, b in zip(point, other)) for other in points) + choices.append((-record["support"], distance, + hashlib.sha256(payload.encode()).hexdigest(), record)) + _, distance, checksum, winner = min(choices) + evidence = {"support": winner["support"], "total_normalized_l1": distance, + "canonical_sha256": checksum} + if candidate["selected"] != winner["options"]: + raise RuntimeError("selected candidate is not the preregistered consensus") + if candidate["selection_evidence"] != evidence: + raise RuntimeError("selection evidence does not reproduce") + + +def mechanism_status(selected): + limit = selected["NULL_LIMIT"] + scoring_off = not limit or selected["NULL_SPAN"] == 0 or selected["NULL_MIN_DEPTH"] >= 30 + fuel_off = not limit or selected["FUEL_NULL"] == 0 or selected["FUEL_MIN_DEPTH"] >= 30 + return { + "qsearch_off": selected["QS"] >= 3000, + "scoring_null_off": scoring_off, + "fuel_null_off": fuel_off, + "all_null_off": scoring_off and fuel_off, + "lmr_off": selected["LMR_LIMIT"] == 0 or selected["LMR_MIN_DEPTH"] >= 30, + "caps_off": selected["FUT_CAP_DEPTH"] < 0, + "iid_on": bool(selected["IID"]), + } + + +def compact_policy(selected, parameters): + """Apply only the preregistered zero-line source rendering.""" + compact = dict(selected) + compact["IID"] = parameters["IID"]["default"] + if not compact["IID"]: + compact["IID_MIN_DEPTH"] = parameters["IID_MIN_DEPTH"]["default"] + compact["IID_RED"] = parameters["IID_RED"]["default"] + return compact + + +def audit(root, space_path): + root, space_path = pathlib.Path(root).resolve(), pathlib.Path(space_path) + verify_seal(root) + candidate = json.loads((root / "candidate.json").read_text()) + space = json.loads(space_path.read_text()) + parameters = {record["name"]: record for record in space["parameters"]} + verify_eventual_space(parameters) + verify_selection(candidate, parameters, space) + selected = candidate["selected"] + if set(selected) != set(parameters): + raise RuntimeError("candidate and parameter-space axes differ") + for name, value in selected.items(): + if value not in domain(parameters[name]): + raise RuntimeError(f"out-of-domain value: {name}={value}") + selected = canonical(selected, space) + limit = selected["NULL_LIMIT"] + compact = compact_policy(selected, parameters) + fuel = compact["FUEL_NULL"] if limit and compact["FUEL_MIN_DEPTH"] < 99 else 0 + lmr = (compact["LMR_RED"] if compact["LMR_LIMIT"] + and compact["LMR_MIN_DEPTH"] < 99 else 0) + defaults = {name: record["default"] for name, record in parameters.items()} + return { + "schema": "sunfish-consensus-prevalidation-audit-v1", + "consensus": str(root), + "consensus_seal_sha256": digest(root / "SEALED"), + "literal": selected, + "compact": compact, + "literal_changes": {name: [defaults[name], value] + for name, value in selected.items() if value != defaults[name]}, + "compact_changes": {name: [defaults[name], value] + for name, value in compact.items() if value != defaults[name]}, + "mechanisms": mechanism_status(selected), + "compact_mechanisms": mechanism_status(compact), + "proof": { + "maximum_real_edge_cost": 1 + fuel + lmr, + "shallow_null_first_depth": selected["NULL_MIN_DEPTH"] + 1, + "shallow_null_last_depth": selected["NULL_MIN_DEPTH"] + selected["NULL_SPAN"], + "positive_null_cap_upper": selected["NULL_LIMIT"] - 1 + + selected["NULL_CAP_MARGIN"], + }, + } + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("consensus", type=pathlib.Path) + parser.add_argument("space", type=pathlib.Path) + args = parser.parse_args() + print(json.dumps(audit(args.consensus, args.space), indent=2, sort_keys=True)) + + +if __name__ == "__main__": + main()