tests/fork_score: 18 tests for v8 ForkScore (#000012 Phase 1a — was zero coverage)
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."
This commit is contained in:
parent
de997f7be3
commit
3d0f02e1ac
1 changed files with 403 additions and 0 deletions
403
tests/test_fork_score.py
Normal file
403
tests/test_fork_score.py
Normal file
|
|
@ -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"
|
||||
Loading…
Add table
Add a link
Reference in a new issue