From 3d0f02e1ac75f7184141f9fea2f0b30a1dc2f5b1 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sun, 10 May 2026 12:38:39 -0400 Subject: [PATCH] =?UTF-8?q?tests/fork=5Fscore:=2018=20tests=20for=20v8=20F?= =?UTF-8?q?orkScore=20(#000012=20Phase=201a=20=E2=80=94=20was=20zero=20cov?= =?UTF-8?q?erage)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit arborist/substrate/fork_score.py landed in #000012 Phase 1a but shipped with no test file. 298 LOC of pure-function scoring + verdict logic, exposed via the `arborist v8 score` CLI (now substrate-rooted per ticket #000035 dir-rename). Coverage: - bench_result_to_metrics adapter (BatteryResult JSON → nested {battery: {sub_battery: metrics}}) - fork_score happy path: pure improvement → ACCEPT (score ≥ SIGNAL_FLOOR=0.05) - marginal band: small improvement → MARGINAL (score in [0, SIGNAL_FLOOR)) - zero parent + zero child → score 0 → MARGINAL - hard-reject paths: per-sub-battery HARD_REGRESSION_FLOOR (≥5pp drop on any 5S/5T/5F sub triggers REJECT regardless of overall positive score) + adaptation_efficiency_neg_infinite_count > 0 → NEG_INF_REGRESSION → REJECT - negative score → REJECT (separate path from hard-reject) - non-bench inputs: capital_delta penalty, audit_completeness bonus, security_risk subtracts WHEN iota>0 (default iota=0 documented) - WeightSet customization flows through to output dict - ScoredFork.to_dict() JSON-serializable - score ≡ Σ breakdown.values() closure (no hidden term) - SIGNAL_FLOOR honored exactly (≥, not >) — score == 0.05 → ACCEPT Fixed-point design discipline: tests use the constants from arborist.substrate.fork_score directly (SIGNAL_FLOOR, HARD_REGRESSION_FLOOR) so a bench-maxing PR that flips the floor forces a tests-fail signal. Default-iota=0 documented explicitly so future readers see "no, you didn't break security_risk; it's deliberately opt-in." --- tests/test_fork_score.py | 403 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 403 insertions(+) create mode 100644 tests/test_fork_score.py diff --git a/tests/test_fork_score.py b/tests/test_fork_score.py new file mode 100644 index 0000000..0731c28 --- /dev/null +++ b/tests/test_fork_score.py @@ -0,0 +1,403 @@ +"""Tests for ``arborist.substrate.fork_score`` — pure-function +ForkScore computation over (parent, child) BatteryResult bundles +(ticket #000012 Phase 1a). + +Covers: + - bench_result_to_metrics adapter (BatteryResult → nested dict) + - fork_score over the three rate-deltas (5S, 5T, 5F) and the + five non-bench inputs + - hard-reject paths: NEG_INF_REGRESSION, per-battery + HARD_REGRESSION_FLOOR + - verdict band: ACCEPT (≥SIGNAL_FLOOR), MARGINAL ([0, SIGNAL_FLOOR)), + REJECT (<0 or hard-reject) + - empty inputs (zero parent, zero child) → score 0, verdict + MARGINAL + - ScoredFork.to_dict serializability + - breakdown sum identity (score ≡ Σ breakdown) +""" + +from __future__ import annotations + +import math + +import pytest + +from arborist.substrate.fork_score import ( + HARD_REGRESSION_FLOOR, + INFINITE_BONUS_CAP, + SIGNAL_FLOOR, + ScoredFork, + bench_result_to_metrics, + fork_score, +) +from arborist.substrate.weights import DEFAULT_WEIGHTS, WeightSet + + +# --- bench_result_to_metrics adapter -------------------------------- + + +def test_bench_result_to_metrics_groups_by_battery(): + """Adapter from runner JSON to nested-dict shape.""" + payload = { + "results": [ + {"battery": "5s", "sub_battery": "syntax", + "metrics": {"parse_pass_rate": 0.9}}, + {"battery": "5s", "sub_battery": "semantics", + "metrics": {"equivalence_recovery_rate": 0.85}}, + {"battery": "5t", "sub_battery": "transfer-learning", + "metrics": {"transfer_learning_success_rate": 0.7}}, + {"battery": "5f", "sub_battery": "function", + "metrics": {"function_pass_rate": 0.95}}, + ], + } + out = bench_result_to_metrics(payload) + assert set(out.keys()) == {"5s", "5t", "5f"} + assert out["5s"]["syntax"]["parse_pass_rate"] == 0.9 + assert out["5s"]["semantics"]["equivalence_recovery_rate"] == 0.85 + assert out["5t"]["transfer-learning"]["transfer_learning_success_rate"] == 0.7 + assert out["5f"]["function"]["function_pass_rate"] == 0.95 + + +def test_bench_result_to_metrics_skips_unknown_battery(): + """Battery keys outside {5s, 5t, 5f} get dropped.""" + payload = { + "results": [ + {"battery": "5r", "sub_battery": "rho", + "metrics": {"rho_rate": 0.5}}, + {"battery": "5s", "sub_battery": "syntax", + "metrics": {"parse_pass_rate": 0.8}}, + ], + } + out = bench_result_to_metrics(payload) + assert "5r" not in out + assert out["5s"]["syntax"]["parse_pass_rate"] == 0.8 + + +def test_bench_result_to_metrics_empty_payload(): + """No `results` key → empty nested dicts (still keyed by battery).""" + out = bench_result_to_metrics({}) + assert out == {"5s": {}, "5t": {}, "5f": {}} + + +# --- fork_score happy path ------------------------------------------ + + +def _bench_dict(rates_by_battery_sub: dict[str, dict[str, dict[str, float]]]): + """Helper: build a nested metrics dict directly.""" + out = {"5s": {}, "5t": {}, "5f": {}} + for battery, subs in rates_by_battery_sub.items(): + out[battery] = {sub: dict(metrics) for sub, metrics in subs.items()} + return out + + +def test_fork_score_pure_improvement_accepts(): + """Child improves on every 5S sub-battery by 10pp → score + ≥ SIGNAL_FLOOR → ACCEPT verdict.""" + parent = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.5}, + "semantics": {"equivalence_recovery_rate": 0.5}, + "syllogism": {"step_validity_rate": 0.5}, + "synthesis": {"derivation_pass_rate": 0.5}, + "semiotics": {"invariance_under_swap": 0.5}, + }, + }) + child = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.6}, + "semantics": {"equivalence_recovery_rate": 0.6}, + "syllogism": {"step_validity_rate": 0.6}, + "synthesis": {"derivation_pass_rate": 0.6}, + "semiotics": {"invariance_under_swap": 0.6}, + }, + }) + r = fork_score(parent, child) + assert r.verdict == "ACCEPT" + assert r.score >= SIGNAL_FLOOR + + +def test_fork_score_marginal_band(): + """Child improves by less than SIGNAL_FLOOR's mapped weight + → MARGINAL band (score in [0, SIGNAL_FLOOR)).""" + # Tiny improvement — 1pp on one sub-battery — well below the + # signal floor when averaged. + parent = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.50}, + "semantics": {"equivalence_recovery_rate": 0.50}, + "syllogism": {"step_validity_rate": 0.50}, + "synthesis": {"derivation_pass_rate": 0.50}, + "semiotics": {"invariance_under_swap": 0.50}, + }, + }) + child = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.51}, + "semantics": {"equivalence_recovery_rate": 0.50}, + "syllogism": {"step_validity_rate": 0.50}, + "synthesis": {"derivation_pass_rate": 0.50}, + "semiotics": {"invariance_under_swap": 0.50}, + }, + }) + r = fork_score(parent, child) + assert r.verdict == "MARGINAL" + assert 0 <= r.score < SIGNAL_FLOOR + + +def test_fork_score_zero_zero_yields_zero(): + """Zero parent + zero child → score 0 (no terms fire) → MARGINAL + (score in [0, SIGNAL_FLOOR)).""" + r = fork_score({"5s": {}, "5t": {}, "5f": {}}, + {"5s": {}, "5t": {}, "5f": {}}) + assert r.score == pytest.approx(0.0, abs=1e-12) + assert r.verdict == "MARGINAL" + + +# --- regression / hard-reject paths --------------------------------- + + +def test_fork_score_hard_regression_5s_rejects(): + """Drop ≥HARD_REGRESSION_FLOOR on a 5S sub → REJECT regardless + of overall positive score.""" + parent = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.9}, + "semantics": {"equivalence_recovery_rate": 0.5}, + "syllogism": {"step_validity_rate": 0.5}, + "synthesis": {"derivation_pass_rate": 0.5}, + "semiotics": {"invariance_under_swap": 0.5}, + }, + }) + # syntax drops by 0.20 (>= 0.05 hard floor); other subs improve. + child = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.7}, + "semantics": {"equivalence_recovery_rate": 0.95}, + "syllogism": {"step_validity_rate": 0.95}, + "synthesis": {"derivation_pass_rate": 0.95}, + "semiotics": {"invariance_under_swap": 0.95}, + }, + }) + r = fork_score(parent, child) + assert r.verdict == "REJECT" + assert any("REGRESSION_5S" in f and "syntax" in f for f in r.flags) + + +def test_fork_score_negative_score_rejects(): + """Score < 0 (more decreases than increases) → REJECT.""" + parent = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.8}, + "semantics": {"equivalence_recovery_rate": 0.8}, + "syllogism": {"step_validity_rate": 0.8}, + "synthesis": {"derivation_pass_rate": 0.8}, + "semiotics": {"invariance_under_swap": 0.8}, + }, + }) + # All decrease by 0.04 (less than HARD_REGRESSION_FLOOR=0.05), + # so no per-sub regression flag — but mean delta is negative + # → score < 0 → REJECT + child = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.76}, + "semantics": {"equivalence_recovery_rate": 0.76}, + "syllogism": {"step_validity_rate": 0.76}, + "synthesis": {"derivation_pass_rate": 0.76}, + "semiotics": {"invariance_under_swap": 0.76}, + }, + }) + r = fork_score(parent, child) + assert r.verdict == "REJECT" + assert r.score < 0 + + +def test_fork_score_neg_infinite_count_in_5f_hard_rejects(): + """`adaptation_efficiency_neg_infinite_count > 0` on child → + auto-REJECT regardless of other terms.""" + parent = _bench_dict({ + "5f": { + "function": {"function_pass_rate": 0.5}, + "finetuning": {"adaptation_improvement_rate": 0.5}, + "falsification": {"error_detection_rate": 0.5}, + "formulate": {"structural_match_rate": 0.5}, + "feedback-loop": {"integration_coverage_rate": 0.5}, + }, + }) + child = _bench_dict({ + "5f": { + "function": {"function_pass_rate": 0.95}, + "finetuning": { + "adaptation_improvement_rate": 0.95, + "adaptation_efficiency_neg_infinite_count": 1, + }, + "falsification": {"error_detection_rate": 0.95}, + "formulate": {"structural_match_rate": 0.95}, + "feedback-loop": {"integration_coverage_rate": 0.95}, + }, + }) + r = fork_score(parent, child) + assert r.verdict == "REJECT" + assert any("NEG_INF_REGRESSION" in f for f in r.flags) + + +# --- non-bench input terms ------------------------------------------ + + +def test_fork_score_capital_cost_penalty_subtracts(): + """capital_delta > 0 reduces the score (cost penalty).""" + parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) + child = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) + no_penalty = fork_score(parent, child, capital_delta=0.0) + with_penalty = fork_score(parent, child, capital_delta=10.0) + # capital_delta term is -theta * capital_delta — so positive + # capital_delta should REDUCE the score + assert with_penalty.score < no_penalty.score + + +def test_fork_score_audit_completeness_adds(): + """audit_completeness > 0 increases the score (positive term).""" + parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) + child = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) + base = fork_score(parent, child) + bonus = fork_score(parent, child, audit_completeness=1.0) + assert bonus.score > base.score + + +def test_fork_score_security_risk_subtracts_when_iota_positive(): + """security_risk reduces the score when the iota weight is + non-zero. Default WeightSet sets iota=0 (security risk is opt- + in for the validator), so we test with an explicit iota>0.""" + parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) + child = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) + iota_on = WeightSet( + alpha=DEFAULT_WEIGHTS.alpha, beta=DEFAULT_WEIGHTS.beta, + gamma=DEFAULT_WEIGHTS.gamma, delta=DEFAULT_WEIGHTS.delta, + epsilon=DEFAULT_WEIGHTS.epsilon, zeta=DEFAULT_WEIGHTS.zeta, + eta=DEFAULT_WEIGHTS.eta, theta=DEFAULT_WEIGHTS.theta, + iota=1.0, # turn on + kappa=DEFAULT_WEIGHTS.kappa, lambda_=DEFAULT_WEIGHTS.lambda_, + ) + base = fork_score(parent, child, weights=iota_on) + risky = fork_score(parent, child, weights=iota_on, security_risk=1.0) + assert risky.score < base.score + + +def test_fork_score_security_risk_inert_under_default_weights(): + """Honest documentation of the default behavior: with iota=0 + (the default), passing security_risk does NOT change the score. + This is by design — operators must opt-in to the security-risk + penalty by setting iota>0.""" + parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) + child = _bench_dict({"5s": {}, "5t": {}, "5f": {}}) + assert DEFAULT_WEIGHTS.iota == 0.0, ( + "test assumes iota=0 default; update if WeightSet defaults change" + ) + base = fork_score(parent, child) + with_risk = fork_score(parent, child, security_risk=1.0) + assert with_risk.score == pytest.approx(base.score, abs=1e-12) + + +# --- weights customization ----------------------------------------- + + +def test_fork_score_weights_recorded_in_output(): + """Custom WeightSet flows through to the output's weights dict.""" + custom = WeightSet(alpha=0.5, beta=0.5, gamma=0.5, + delta=0.1, epsilon=0.1, zeta=0.1, + eta=1.0, theta=1.0, iota=1.0, + kappa=1.0, lambda_=1.0) + r = fork_score({}, {}, weights=custom) + assert r.weights["alpha"] == pytest.approx(0.5) + assert r.weights["lambda_"] == pytest.approx(1.0) + + +def test_fork_score_default_weights_used_when_unspecified(): + r = fork_score({}, {}) + assert r.weights == DEFAULT_WEIGHTS.as_dict() + + +# --- ScoredFork API ------------------------------------------------- + + +def test_scored_fork_to_dict_serializable(): + """ScoredFork.to_dict() returns JSON-serializable types.""" + r = fork_score({}, {}) + d = r.to_dict() + import json + + json.dumps(d) # must not raise + assert "score" in d + assert "verdict" in d + assert "breakdown" in d + assert "flags" in d + assert "weights" in d + + +# --- score = sum(breakdown) closure --------------------------------- + + +def test_score_equals_sum_of_breakdown(): + """Closure check: score ≡ Σ breakdown values. Multiple + parent/child configurations to widen the cone.""" + configs = [ + # Plain improvement on 5S only. + ( + {"5s": {"syntax": {"parse_pass_rate": 0.5}}}, + {"5s": {"syntax": {"parse_pass_rate": 0.7}}}, + ), + # Mixed deltas across all three batteries. + ( + _bench_dict({ + "5s": {"syntax": {"parse_pass_rate": 0.6}}, + "5t": {"transfer-learning": {"transfer_learning_success_rate": 0.6}}, + "5f": {"function": {"function_pass_rate": 0.6}}, + }), + _bench_dict({ + "5s": {"syntax": {"parse_pass_rate": 0.7}}, + "5t": {"transfer-learning": {"transfer_learning_success_rate": 0.65}}, + "5f": {"function": {"function_pass_rate": 0.55}}, + }), + ), + # All-zero inputs → score 0. + ({"5s": {}, "5t": {}, "5f": {}}, {"5s": {}, "5t": {}, "5f": {}}), + ] + for parent, child in configs: + r = fork_score(parent, child) + assert r.score == pytest.approx(sum(r.breakdown.values()), abs=1e-9), ( + f"score {r.score} ≠ Σbreakdown {sum(r.breakdown.values())} on " + f"parent={parent!r}, child={child!r}" + ) + + +# --- constants honored from arborist.substrate.fork_score ----------- + + +def test_signal_floor_honored(): + """Score exactly at SIGNAL_FLOOR is ACCEPT (≥, not >).""" + # We construct a child whose 5S delta * alpha == SIGNAL_FLOOR. + # alpha defaults to 0.30 per WeightSet (see arborist.substrate.weights). + # Solve for delta: delta = SIGNAL_FLOOR / alpha = 0.05 / 0.30 ≈ 0.1667. + # Average of 5 sub-deltas; set each sub to 0.1667 to hit it. + parent = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.5}, + "semantics": {"equivalence_recovery_rate": 0.5}, + "syllogism": {"step_validity_rate": 0.5}, + "synthesis": {"derivation_pass_rate": 0.5}, + "semiotics": {"invariance_under_swap": 0.5}, + }, + }) + target_delta = SIGNAL_FLOOR / DEFAULT_WEIGHTS.alpha + child = _bench_dict({ + "5s": { + "syntax": {"parse_pass_rate": 0.5 + target_delta}, + "semantics": {"equivalence_recovery_rate": 0.5 + target_delta}, + "syllogism": {"step_validity_rate": 0.5 + target_delta}, + "synthesis": {"derivation_pass_rate": 0.5 + target_delta}, + "semiotics": {"invariance_under_swap": 0.5 + target_delta}, + }, + }) + r = fork_score(parent, child) + # Score = alpha · target_delta = SIGNAL_FLOOR exactly. Verdict ACCEPT. + assert r.score == pytest.approx(SIGNAL_FLOOR, abs=1e-9) + assert r.verdict == "ACCEPT"