File size: 6,890 Bytes
4be6a52 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | import copy
import math
from dataclasses import asdict
import pytest
from stackcraft.engine import new_game
from stackcraft.evaluation import paired_bootstrap, paired_report, summarize_positions
from stackcraft.players import Decision, observe
from stackcraft.schema import RULES_VERSION
def episode(player: str, seed: int, lines: int, *, failed: bool = False) -> dict:
"""Statistical fixture, not a claimed simulator run."""
errors = [{"turn": 10, "kind": "player_error", "message": "toy failure"}] if failed else []
return {
"schema_version": 1,
"rules_version": RULES_VERSION,
"player_id": player,
"player": {"name": player, "revision": "toy-v1", "runtime_config": {}},
"seed": seed,
"max_pieces": 100,
"outcome": {
"score": 100 * lines,
"lines": lines,
"pieces": 10,
"terminal": not failed,
"cap_hit": False,
"status": "error" if failed else "top_out",
},
"errors": errors,
"decisions": [{"decision_seconds": 0.01, "input_tokens": 100}] * (10 + failed),
}
def report(episodes, **kwargs):
return paired_report(
episodes, trained_id="trained", base_id="base", bootstrap_samples=200, **kwargs
)
def test_paired_constant_improvement_has_exact_interval() -> None:
result = paired_bootstrap([3, 3, 3], samples=200)
assert result["mean_difference"] == result["ci95_lower"] == result["ci95_upper"] == 3
assert result == paired_bootstrap([3, 3, 3], samples=200)
def test_pairing_preserves_difficulty_and_is_order_independent() -> None:
episodes = [
episode("base", 80, 30),
episode("trained", 80, 33),
episode("base", 70, 1),
episode("trained", 70, 4),
]
result = report(episodes)
assert result == report(episodes[::-1])
assert result["seeds"] == [70, 80]
assert result["paired_trained_minus_base"]["lines"]["ci95_lower"] == 3
assert result["positive_mean_lines_signal"]
assert not result["selection_performed"]
assert not result["matches_reserved_test_pool"]
assert result["players"]["trained"]["latency_seconds"]["p95"] == 0.01
assert result["players"]["trained"]["input_tokens"]["mean"] == 100
def test_failure_is_counted_as_zero_and_partial_outcome_is_retained() -> None:
result = report([episode("base", 70, 2), episode("trained", 70, 50, failed=True)])
assert result["players"]["trained"]["episodes"] == 1
assert result["players"]["trained"]["failure_adjusted"]["lines"]["mean"] == 0
assert result["players"]["trained"]["observed_before_error"]["lines"]["mean"] == 50
assert result["players"]["trained"]["error_rate"] == 1
assert result["players"]["trained"]["failed_seeds"] == [70]
assert result["paired_trained_minus_base"]["lines"]["mean_difference"] == -2
assert not result["positive_mean_lines_signal"]
def test_positive_interval_cannot_hide_base_failures() -> None:
rows = [
episode(player, seed, 2, failed=player == "base")
for player in ("base", "trained")
for seed in (70, 80)
]
result = report(rows)
assert result["paired_trained_minus_base"]["lines"]["ci95_lower"] == 2
assert not result["errors_acceptable"]
assert not result["positive_mean_lines_signal"]
@pytest.mark.parametrize(
"mutation",
[
"duplicate",
"missing",
"cap",
"rules",
"revision",
"runtime",
"bad_metric",
"bad_duration",
"status",
],
)
def test_mismatched_or_invalid_episodes_are_rejected(mutation: str) -> None:
rows = [episode(player, seed, 2) for player in ("base", "trained") for seed in (70, 80)]
if mutation == "duplicate":
rows.append(copy.deepcopy(rows[0]))
elif mutation == "missing":
rows.pop()
elif mutation == "cap":
rows[-1]["max_pieces"] = 200
elif mutation == "rules":
rows[-1]["rules_version"] = "other"
elif mutation == "revision":
rows[-1]["player"]["revision"] = "later-checkpoint"
elif mutation == "runtime":
rows[-1]["player"]["runtime_config"] = {"max_length": 12}
elif mutation == "bad_metric":
rows[-1]["outcome"]["lines"] = float("nan")
elif mutation == "bad_duration":
rows[-1]["decisions"][0]["decision_seconds"] = -1
else:
rows[-1]["outcome"]["cap_hit"] = True
with pytest.raises(ValueError):
report(rows)
def positions():
observation = observe(new_game(70))
target = observation.legal_actions[0].id
records = [
{"id": str(index), "action_id": target, "observation": asdict(observation)}
for index in range(2)
]
probabilities = {a.id: 1 / len(observation.legal_actions) for a in observation.legal_actions}
return records, target, probabilities
def test_position_metrics_are_separate_and_probability_scores_correct() -> None:
records, target, probabilities = positions()
result = summarize_positions(
records, {row["id"]: Decision(target, probabilities) for row in records}
)
assert result["teacher_agreement"] == 1
assert result["mean_nll"] == pytest.approx(math.log(len(probabilities)))
assert result["mean_brier"] == pytest.approx(1 - 1 / len(probabilities))
assert result["complete_probability_coverage"]
assert not result["selection_performed"]
def test_missing_predictions_remain_in_agreement_denominator() -> None:
records, target, probabilities = positions()
result = summarize_positions(records, {"0": Decision(target, probabilities)})
assert result["teacher_agreement"] == 0.5
assert result["error_rate"] == 0.5
assert result["failed_predictions"] == ["1"]
assert result["mean_nll"] is None and result["mean_brier"] is None
assert result["finite_subset_nll_positions"] == 1
def test_missing_probabilities_and_zero_probability_are_explicit() -> None:
records, target, probabilities = positions()
result = summarize_positions(records, {row["id"]: Decision(target) for row in records})
assert result["teacher_agreement"] == 1
assert result["mean_nll"] is None
assert result["missing_probabilities"] == ["0", "1"]
for action in probabilities:
probabilities[action] = 0 if action == target else 1 / (len(probabilities) - 1)
result = summarize_positions(
records, {row["id"]: Decision(target, probabilities) for row in records}
)
assert result["nll_is_infinite"]
assert result["zero_target_probability_count"] == 2
assert result["mean_nll"] is None
def test_degenerate_single_episode_cannot_claim_positive_signal() -> None:
result = report([episode("base", 70, 1), episode("trained", 70, 2)])
assert result["paired_trained_minus_base"]["lines"]["ci95_lower"] == 1
assert not result["positive_mean_lines_signal"]
|