"One more iteration then close" (fox): added committed KAT-regeneration
scripts for both the T3 calculator and φ_PRG — the regen step was a
throwaway temp script before; now it's reproducible and the phi_prg
test's skipif reason ("run scripts/generate_phi_prg_kat.py") points at
a file that exists. Then closed #000036.
New scripts:
- scripts/generate_t3_bound_kat.py — regenerates
bench/fixtures/t3-bound/known-answer-tests.jsonl from a fixed 12-config
list (the §7 worked examples under max_envelope + non-default-C_B*
+ g=0 edge + explicit-b1_model pins for the other three models).
- scripts/generate_phi_prg_kat.py — regenerates
bench/fixtures/phi-prg/known-answer-tests.jsonl from a fixed 10-entry
list (placeholder/random seeds, one-bit-flip variants, block-boundary
dim_h=16/17, 4096 counter-rollover stress).
- Both verified to reproduce the committed fixture data lines byte-
for-byte (only the header comments changed, to reference the script).
Each docstring states: run after any algorithm change, then bump the
module version (CALCULATOR_VERSION / PHI_PRG_VERSION) so the fixture's
version field changes too.
Doc/test:
- test_t3_bound_calculator.py skipif reason now references the regen
script (matches the phi_prg test pattern).
- #000035 §3.3 + t3-bound.md §10.1 reference the regen scripts.
Closure (#000036):
- Status → closed · 2026-05-11 in the ticket file + TICKETS.md row.
Phase 1 + dav1d Tier-1/Tier-2 (Option B in v1) + KAT-regen tooling
all landed; all §5 acceptance criteria met; both dav1d closure
blockers cleared. Continuation: empirical C_B1/C_B2/C_B3 tightening
under #000043 (parks on v7 deployment data); landing the bound's
framing into a v7 plastic-training spec parks on that spec gaining
a deployment target; R2's architectural integrations (Merkle audit-
event commitment, SQD canonicalization, CTI clause-lattice, 5F
trigger, ForkScore security-risk) are separate tickets if wanted.
- t3-bound.md header flipped to "closed 2026-05-11".
Full suite: 2312 passed, 28 skipped.
648 lines
26 KiB
Python
648 lines
26 KiB
Python
"""Tests for the T3 per-window covert-channel bound calculator
|
||
(`bench/scripts/t3_bound_calculator.py`, ticket #000036 §11).
|
||
|
||
Validates the closed-form bound from `docs/soft-hash-channel-t3-
|
||
bound.md` §6 against:
|
||
|
||
- mathematical identities (B1 / B2 / B3 isolation, monotonicity in
|
||
each input)
|
||
- input-validation hard checks
|
||
- operator-guidance text mode transitions (≤0 / <256 / ≥256 bits)
|
||
- CLI surface (argparse + JSON output, error path)
|
||
- the §11 worked-example numbers in the ticket doc
|
||
|
||
Pure stdlib — same dep profile as the calculator itself.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import pathlib
|
||
import subprocess
|
||
import sys
|
||
|
||
import pytest
|
||
|
||
from bench.scripts.t3_bound_calculator import (
|
||
CALCULATOR_VERSION,
|
||
t3_bound_bits,
|
||
)
|
||
|
||
|
||
# Repo root for CLI subprocess tests — derived, not hard-coded, so the
|
||
# suite runs on any checkout (dav1d review 2026-05-11).
|
||
REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||
|
||
|
||
# --- baseline (§11 worked example) -----------------------------------
|
||
|
||
|
||
_BASELINE = dict(
|
||
gradient_fraction=0.05,
|
||
gradient_norm_max=1.0,
|
||
gradient_noise_stddev=0.1,
|
||
lr_decision_interval=100,
|
||
lr_grid_size=8,
|
||
window_length=10000,
|
||
batches_per_epoch=1024,
|
||
steps_per_epoch=1024,
|
||
)
|
||
|
||
|
||
def test_baseline_matches_section_11_doc():
|
||
"""The §11 worked example baseline under the default b1_model=
|
||
"max_envelope" (dav1d review 2026-05-11). Calculator's actual numbers:
|
||
|
||
I_window_bits_upper_bound: 6183.0154
|
||
B1_contribution: 5849.625 (aggregate_bias selected)
|
||
B1_fraction_channels: 1729.7158
|
||
B1_aggregate_bias: 5849.625
|
||
B1_effective_control_v1: 292.4813 (the older non-worst-case)
|
||
B2_contribution: 300.0
|
||
B3_contribution: 33.3904
|
||
snr_grad: 0.5 (g·G/σ)
|
||
snr_per_channel: 10.0 (G/σ)
|
||
|
||
The conservative envelope takes max(fraction_channels, aggregate_bias);
|
||
at this baseline aggregate_bias (5849.6) dominates fraction_channels
|
||
(1729.7). The old effective_control formula (292.5) is still reported
|
||
as B1_effective_control_v1 for comparison.
|
||
"""
|
||
r = t3_bound_bits(**_BASELINE)
|
||
assert r["b1_model"] == "max_envelope"
|
||
assert r["b1_selected"] == "aggregate_bias"
|
||
# Total: ⟨B1+B2+B3⟩ ≈ 5849.63 + 300.0 + 33.39 = 6183.02
|
||
assert r["I_window_bits_upper_bound"] == pytest.approx(6183.02, abs=0.05)
|
||
# B2 is exact: ⌈10000/100⌉ · log₂(8) = 100 · 3 = 300
|
||
assert r["B2_contribution"] == pytest.approx(300.0, abs=1e-3)
|
||
# B3: epochs = ⌈10000/1024⌉ = 10; per-epoch factor = 1024·0.1/1.0 = 102.4;
|
||
# log₂(102.4)/2 ≈ 3.339; total ≈ 33.39 bits.
|
||
assert r["B3_contribution"] == pytest.approx(33.39, abs=0.05)
|
||
# B1 (max_envelope): aggregate_bias = W·log₂(1 + g·G/σ) = 10000·log₂(1.5)
|
||
# ≈ 10000·0.585 = 5849.63 (> fraction_channels 1729.72).
|
||
assert r["B1_contribution"] == pytest.approx(5849.63, abs=0.05)
|
||
assert r["B1_aggregate_bias"] == pytest.approx(5849.63, abs=0.05)
|
||
# fraction_channels = g·W·log₂(1 + G/σ) = 0.05·10000·log₂(11) ≈ 1729.72
|
||
assert r["B1_fraction_channels"] == pytest.approx(1729.72, abs=0.05)
|
||
# effective_control_v1 (old formula) still reported: g·W·log₂(1+g·G/σ)
|
||
# = 0.05·10000·log₂(1.5) ≈ 292.48
|
||
assert r["B1_effective_control_v1"] == pytest.approx(292.48, abs=0.05)
|
||
assert r["snr_grad"] == pytest.approx(0.5, abs=1e-6) # g·G/σ
|
||
assert r["snr_per_channel"] == pytest.approx(10.0, abs=1e-6) # G/σ
|
||
|
||
|
||
def test_returns_calculator_version_token():
|
||
r = t3_bound_bits(**_BASELINE)
|
||
assert r["calculator_version"] == CALCULATOR_VERSION
|
||
# versioned-default discipline: matches the v1-bottou-refinement token
|
||
assert "v1" in CALCULATOR_VERSION
|
||
|
||
|
||
def test_b2_isolated_zero_when_decisions_zero():
|
||
"""If lr_decision_interval > window_length, decisions_in_window = 1
|
||
(one initial decision); the bound floor is log₂ R, not zero.
|
||
Exercises the ceil semantics on small W / large K.
|
||
"""
|
||
r = t3_bound_bits(**{**_BASELINE, "window_length": 1, "lr_decision_interval": 100})
|
||
# ceil(1/100) = 1, so B2 = log₂ 8 = 3.0
|
||
assert r["B2_contribution"] == pytest.approx(3.0, abs=1e-6)
|
||
|
||
|
||
def test_snr_grad_formula():
|
||
"""SNR_grad = g · ‖∇L_max‖ / σ_grad. Check directly."""
|
||
r = t3_bound_bits(**_BASELINE)
|
||
expected_snr = 0.05 * 1.0 / 0.1 # 0.5
|
||
assert r["snr_grad"] == pytest.approx(expected_snr, abs=1e-6)
|
||
|
||
|
||
# --- monotonicity / bound shape --------------------------------------
|
||
|
||
|
||
def test_monotone_in_window_length():
|
||
"""Doubling W roughly doubles each contribution (B1 linear in W,
|
||
B2 + B3 grow with ⌈W/K⌉ + ⌈W/E⌉ which scale linearly with W on
|
||
aligned periods)."""
|
||
base = t3_bound_bits(**_BASELINE)
|
||
long_window = t3_bound_bits(**{**_BASELINE, "window_length": 20000})
|
||
# Total grows monotonically.
|
||
assert long_window["I_window_bits_upper_bound"] > base["I_window_bits_upper_bound"]
|
||
# B1 is exactly linear in W, so doubling W doubles B1.
|
||
assert long_window["B1_contribution"] == pytest.approx(
|
||
2 * base["B1_contribution"], rel=1e-6
|
||
)
|
||
|
||
|
||
def test_monotone_in_gradient_fraction():
|
||
"""Increasing g raises the bound (more adversarial control)."""
|
||
base = t3_bound_bits(**_BASELINE)
|
||
high_g = t3_bound_bits(**{**_BASELINE, "gradient_fraction": 0.5})
|
||
assert high_g["I_window_bits_upper_bound"] > base["I_window_bits_upper_bound"]
|
||
# B2 + B3 do not depend on g, so the increase is in B1 only.
|
||
assert high_g["B2_contribution"] == base["B2_contribution"]
|
||
assert high_g["B3_contribution"] == base["B3_contribution"]
|
||
|
||
|
||
def test_b3_floor_at_zero_when_factor_below_one():
|
||
"""When N_b · σ_grad / ‖∇L_max‖ ≤ 1, log₂ goes ≤ 0; the bound
|
||
floors at 0 — adversary can't do worse than random shuffle in
|
||
this regime. Verify the floor."""
|
||
# σ_grad = 0.001, gradient_norm_max = 100, N_b = 4 → 0.00004 < 1
|
||
args = {**_BASELINE, "gradient_noise_stddev": 0.001,
|
||
"gradient_norm_max": 100.0, "batches_per_epoch": 4,
|
||
"steps_per_epoch": 4}
|
||
r = t3_bound_bits(**args)
|
||
assert r["B3_contribution"] == pytest.approx(0.0, abs=1e-9)
|
||
|
||
|
||
def test_constant_C_B1_scales_b1():
|
||
"""C_B1 · B1 — halving C_B1 halves B1 (calculator rounds output
|
||
to 4 dp; tolerate rounding)."""
|
||
base = t3_bound_bits(**_BASELINE)
|
||
half_c = t3_bound_bits(**_BASELINE, c_b1=0.5)
|
||
assert half_c["B1_contribution"] == pytest.approx(
|
||
0.5 * base["B1_contribution"], abs=1e-3
|
||
)
|
||
# B2 / B3 unchanged.
|
||
assert half_c["B2_contribution"] == base["B2_contribution"]
|
||
assert half_c["B3_contribution"] == base["B3_contribution"]
|
||
|
||
|
||
def test_constants_recorded():
|
||
r = t3_bound_bits(**_BASELINE, c_b1=0.4, c_b2=0.3, c_b3=0.2)
|
||
assert r["constants"] == {"C_B1": 0.4, "C_B2": 0.3, "C_B3": 0.2}
|
||
|
||
|
||
# --- input validation ------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize("g", [-0.1, 1.5, 2.0])
|
||
def test_invalid_gradient_fraction_rejected(g):
|
||
args = {**_BASELINE, "gradient_fraction": g}
|
||
with pytest.raises(ValueError, match="gradient_fraction"):
|
||
t3_bound_bits(**args)
|
||
|
||
|
||
def test_gradient_fraction_zero_accepted():
|
||
"""g = 0 means no T2 gradient surface — B1 = 0, but T3's LR +
|
||
batch-order channels (B2, B3) still fire. Component-isolation case
|
||
(dav1d review 2026-05-11: allow 0 ≤ g ≤ 1)."""
|
||
r = t3_bound_bits(**{**_BASELINE, "gradient_fraction": 0.0})
|
||
assert r["B1_contribution"] == pytest.approx(0.0, abs=1e-9)
|
||
assert r["snr_grad"] == pytest.approx(0.0, abs=1e-9)
|
||
assert r["B2_contribution"] == pytest.approx(300.0, abs=1e-3)
|
||
assert r["I_window_bits_upper_bound"] == pytest.approx(
|
||
r["B2_contribution"] + r["B3_contribution"], abs=1e-3
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("name,bad", [
|
||
("window_length", True),
|
||
("window_length", False),
|
||
("lr_grid_size", True),
|
||
("batches_per_epoch", False),
|
||
("lr_decision_interval", True),
|
||
("steps_per_epoch", True),
|
||
])
|
||
def test_bool_rejected_for_int_fields(name, bad):
|
||
"""Python's isinstance(True, int) is True — a bare isinstance check
|
||
lets bools leak into a security calculation. Reject explicitly via
|
||
type(value) is int (dav1d review 2026-05-11)."""
|
||
with pytest.raises(ValueError, match=name):
|
||
t3_bound_bits(**{**_BASELINE, name: bad})
|
||
|
||
|
||
@pytest.mark.parametrize("name,bad", [
|
||
("gradient_fraction", True),
|
||
("gradient_norm_max", False),
|
||
("gradient_noise_stddev", True),
|
||
("c_b1", True),
|
||
("c_b2", False),
|
||
])
|
||
def test_bool_rejected_for_float_fields(name, bad):
|
||
with pytest.raises(ValueError, match=name):
|
||
t3_bound_bits(**{**_BASELINE, name: bad})
|
||
|
||
|
||
@pytest.mark.parametrize("name,bad", [
|
||
("gradient_fraction", float("nan")),
|
||
("gradient_fraction", float("inf")),
|
||
("gradient_norm_max", float("nan")),
|
||
("gradient_norm_max", float("inf")),
|
||
("gradient_noise_stddev", float("inf")),
|
||
("gradient_noise_stddev", float("nan")),
|
||
("c_b1", float("nan")),
|
||
("c_b2", float("inf")),
|
||
("c_b3", float("-inf")),
|
||
])
|
||
def test_nonfinite_numbers_rejected(name, bad):
|
||
"""NaN / ±inf in any numeric field → ValueError, not a poisoned
|
||
bound that emits inf in the recommendation (dav1d review 2026-05-11)."""
|
||
with pytest.raises(ValueError, match=name):
|
||
t3_bound_bits(**{**_BASELINE, name: bad})
|
||
|
||
|
||
@pytest.mark.parametrize("name,value", [
|
||
("gradient_norm_max", 0.0),
|
||
("gradient_norm_max", -1.0),
|
||
("gradient_noise_stddev", 0.0),
|
||
("gradient_noise_stddev", -0.1),
|
||
])
|
||
def test_invalid_positive_floats_rejected(name, value):
|
||
args = {**_BASELINE, name: value}
|
||
with pytest.raises(ValueError, match=name):
|
||
t3_bound_bits(**args)
|
||
|
||
|
||
@pytest.mark.parametrize("name", [
|
||
"lr_decision_interval", "lr_grid_size", "window_length",
|
||
"batches_per_epoch", "steps_per_epoch",
|
||
])
|
||
@pytest.mark.parametrize("bad", [0, -1, 1.5, "a"])
|
||
def test_invalid_positive_int_rejected(name, bad):
|
||
args = {**_BASELINE, name: bad}
|
||
with pytest.raises(ValueError, match=name):
|
||
t3_bound_bits(**args)
|
||
|
||
|
||
@pytest.mark.parametrize("c_name,c_val", [
|
||
("c_b1", -0.1), ("c_b1", 1.5),
|
||
("c_b2", -0.5), ("c_b2", 2.0),
|
||
("c_b3", -1.0), ("c_b3", 1.001),
|
||
])
|
||
def test_constants_clamped_to_unit_interval(c_name, c_val):
|
||
args = {**_BASELINE, c_name: c_val}
|
||
with pytest.raises(ValueError, match=c_name):
|
||
t3_bound_bits(**args)
|
||
|
||
|
||
# --- recommendation text mode transitions ----------------------------
|
||
|
||
|
||
def test_recommendation_total_le_zero():
|
||
"""When the bound floors at 0 across all three terms (extreme
|
||
benign config), recommendation reads as 'vacuously safe'."""
|
||
args = {
|
||
**_BASELINE,
|
||
"gradient_fraction": 1e-9, # near-zero g → tiny B1
|
||
"lr_decision_interval": 1_000_000, # huge K → ⌈W/K⌉ = 1, but log₂ R = log₂ 1 = 0 below
|
||
"lr_grid_size": 1, # log₂(1) = 0 → B2 = 0
|
||
"gradient_noise_stddev": 0.001,
|
||
"gradient_norm_max": 100.0,
|
||
"batches_per_epoch": 4,
|
||
"steps_per_epoch": 4, # b3_factor = 4·0.001/100 = 4e-5 < 1 → B3 floor
|
||
}
|
||
r = t3_bound_bits(**args)
|
||
assert r["I_window_bits_upper_bound"] >= 0
|
||
# B1 is tiny but not exactly 0; total floor regime depends on
|
||
# whether the sum reaches positive territory. Assert at least
|
||
# B2 + B3 are zero on this config.
|
||
assert r["B2_contribution"] == pytest.approx(0.0, abs=1e-9)
|
||
assert r["B3_contribution"] == pytest.approx(0.0, abs=1e-9)
|
||
|
||
|
||
def test_recommendation_below_sha256():
|
||
"""The default-baseline bound exceeds 256 under max_envelope; use a
|
||
small W that lands in the (0, 256) certified band. At W=300:
|
||
aggregate_bias ≈ 175.5, B2 = 9, B3 ≈ 3.3 → total ≈ 187.8."""
|
||
args = {**_BASELINE, "window_length": 300}
|
||
r = t3_bound_bits(**args)
|
||
assert 0 < r["I_window_bits_upper_bound"] < 256, (
|
||
f"window_length=300 should land in (0, 256); got {r['I_window_bits_upper_bound']}"
|
||
)
|
||
assert "windows" in r["recommendation"]
|
||
assert "256" in r["recommendation"]
|
||
assert r["certification_status"] == "CERTIFIED_BY_BOUND"
|
||
|
||
|
||
def test_recommendation_exceeds_sha256():
|
||
"""Baseline (W=10000) → ~626 bits > 256 — recommendation flags that
|
||
the conservative bound CANNOT CERTIFY the residual (not 'guarantee
|
||
broken'; dav1d review 2026-05-11: I_window is an upper bound, so
|
||
exceeding 256 means we cannot certify, not that the adversary can
|
||
steer 256 bits)."""
|
||
r = t3_bound_bits(**_BASELINE)
|
||
assert r["I_window_bits_upper_bound"] > 256
|
||
assert "EXCEEDS" in r["recommendation"]
|
||
assert "CANNOT CERTIFY" in r["recommendation"]
|
||
assert "256" in r["recommendation"]
|
||
assert r["certification_status"] == "NOT_CERTIFIED_BY_BOUND"
|
||
|
||
|
||
def test_output_carries_b1_model_and_certification_fields():
|
||
"""Structured machine-readable fields (dav1d review 2026-05-11):
|
||
b1_model, b1_selected, certification_status,
|
||
certification_threshold_bits, model_assumptions, plus all three
|
||
B1 model variants + both SNR readings."""
|
||
r = t3_bound_bits(**_BASELINE)
|
||
assert r["b1_model"] == "max_envelope" # the new default
|
||
assert r["b1_selected"] in {"fraction_channels", "aggregate_bias"}
|
||
assert r["certification_threshold_bits"] == 256
|
||
assert r["certification_status"] in {
|
||
"CERTIFIED_BY_BOUND", "NOT_CERTIFIED_BY_BOUND"
|
||
}
|
||
assert "B1_model_max_envelope" in r["model_assumptions"]
|
||
# all three B1 variants reported regardless of which is selected
|
||
for k in ("B1_fraction_channels", "B1_aggregate_bias",
|
||
"B1_effective_control_v1", "snr_per_channel"):
|
||
assert k in r, f"missing output field {k!r}"
|
||
assert r["certification_status"] == "NOT_CERTIFIED_BY_BOUND" # baseline > 256
|
||
|
||
|
||
def test_certification_status_certified_below_threshold():
|
||
"""A config landing in (0, 256) → CERTIFIED_BY_BOUND. W=300 under
|
||
max_envelope yields ~188 bits."""
|
||
r = t3_bound_bits(**{**_BASELINE, "window_length": 300})
|
||
assert 0 < r["I_window_bits_upper_bound"] < 256
|
||
assert r["certification_status"] == "CERTIFIED_BY_BOUND"
|
||
|
||
|
||
# --- CLI surface (subprocess invocation) -----------------------------
|
||
|
||
|
||
def test_cli_baseline_runs_clean(tmp_path):
|
||
"""Invoke the calculator CLI with §11 worked-example flags;
|
||
parse the JSON output; verify the I_window_bits_upper_bound
|
||
matches the function call."""
|
||
cmd = [
|
||
sys.executable, "-m", "bench.scripts.t3_bound_calculator",
|
||
"--gradient-fraction", "0.05",
|
||
"--gradient-norm-max", "1.0",
|
||
"--gradient-noise-stddev", "0.1",
|
||
"--lr-decision-interval", "100",
|
||
"--lr-grid-size", "8",
|
||
"--window-length", "10000",
|
||
"--batches-per-epoch", "1024",
|
||
"--steps-per-epoch", "1024",
|
||
]
|
||
out = subprocess.run(cmd, capture_output=True, text=True, check=True,
|
||
cwd=REPO_ROOT)
|
||
j = json.loads(out.stdout)
|
||
assert j["calculator_version"] == CALCULATOR_VERSION
|
||
assert j["b1_model"] == "max_envelope" # CLI default
|
||
# Closed-form actual under max_envelope: 5849.63 + 300.0 + 33.39 = 6183.02
|
||
assert j["I_window_bits_upper_bound"] == pytest.approx(6183.02, abs=0.05)
|
||
|
||
|
||
def test_cli_b1_model_flag():
|
||
"""--b1-model effective_control_v1 reverts to the older formula;
|
||
--b1-model fraction_channels picks that envelope term."""
|
||
base_cmd = [
|
||
sys.executable, "-m", "bench.scripts.t3_bound_calculator",
|
||
"--gradient-fraction", "0.05", "--gradient-norm-max", "1.0",
|
||
"--gradient-noise-stddev", "0.1", "--lr-decision-interval", "100",
|
||
"--lr-grid-size", "8", "--window-length", "10000",
|
||
"--batches-per-epoch", "1024", "--steps-per-epoch", "1024",
|
||
]
|
||
for model, expected_b1 in [
|
||
("effective_control_v1", 292.48),
|
||
("fraction_channels", 1729.72),
|
||
("aggregate_bias", 5849.63),
|
||
]:
|
||
out = subprocess.run(base_cmd + ["--b1-model", model],
|
||
capture_output=True, text=True, check=True,
|
||
cwd=REPO_ROOT)
|
||
j = json.loads(out.stdout)
|
||
assert j["b1_model"] == model
|
||
assert j["B1_contribution"] == pytest.approx(expected_b1, abs=0.05), (
|
||
f"b1_model={model}: B1={j['B1_contribution']} vs expected {expected_b1}"
|
||
)
|
||
|
||
|
||
def test_cli_invalid_input_exits_2():
|
||
"""ValueError → exit code 2 + stderr error message."""
|
||
cmd = [
|
||
sys.executable, "-m", "bench.scripts.t3_bound_calculator",
|
||
"--gradient-fraction", "1.5", # invalid
|
||
"--gradient-norm-max", "1.0",
|
||
"--gradient-noise-stddev", "0.1",
|
||
"--lr-decision-interval", "100",
|
||
"--lr-grid-size", "8",
|
||
"--window-length", "100",
|
||
"--batches-per-epoch", "1024",
|
||
"--steps-per-epoch", "1024",
|
||
]
|
||
out = subprocess.run(cmd, capture_output=True, text=True,
|
||
cwd=REPO_ROOT)
|
||
assert out.returncode == 2
|
||
assert "gradient_fraction" in out.stderr
|
||
|
||
|
||
# --- closed-form sanity ---------------------------------------------
|
||
|
||
|
||
def test_b2_exact_formula():
|
||
"""B2 = C_B2 · ⌈W/K⌉ · log₂(R). Direct closed-form check
|
||
against several configurations."""
|
||
cases = [
|
||
# (W, K, R) → expected ⌈W/K⌉ · log₂ R
|
||
(1000, 100, 8, 10 * 3.0),
|
||
(1000, 200, 4, 5 * 2.0),
|
||
(999, 100, 16, 10 * 4.0), # ceil(999/100)=10
|
||
(1, 100, 2, 1 * 1.0),
|
||
]
|
||
for W, K, R, expected in cases:
|
||
r = t3_bound_bits(
|
||
**{**_BASELINE, "window_length": W,
|
||
"lr_decision_interval": K, "lr_grid_size": R},
|
||
)
|
||
assert r["B2_contribution"] == pytest.approx(expected, abs=1e-9), (
|
||
f"B2 mismatch on (W={W}, K={K}, R={R}): "
|
||
f"{r['B2_contribution']} vs expected {expected}"
|
||
)
|
||
|
||
|
||
def test_b1_effective_control_v1_exact_formula():
|
||
"""b1_model="effective_control_v1": B1 = C_B1 · g · W · log₂(SNR+1)
|
||
where SNR = g·‖∇L_max‖/σ_grad (the older non-worst-case formula).
|
||
Hand-compute for one config and check exact agreement."""
|
||
g, gnm, sigma, W = 0.1, 2.0, 0.5, 5000
|
||
snr = g * gnm / sigma # 0.4
|
||
expected = 1.0 * g * W * math.log2(snr + 1)
|
||
r = t3_bound_bits(**{
|
||
**_BASELINE,
|
||
"gradient_fraction": g, "gradient_norm_max": gnm,
|
||
"gradient_noise_stddev": sigma, "window_length": W,
|
||
"b1_model": "effective_control_v1",
|
||
})
|
||
assert r["b1_model"] == "effective_control_v1"
|
||
assert r["b1_selected"] == "effective_control_v1"
|
||
assert r["B1_contribution"] == pytest.approx(expected, abs=1e-3)
|
||
assert r["B1_effective_control_v1"] == pytest.approx(expected, abs=1e-3)
|
||
assert r["snr_grad"] == pytest.approx(snr, abs=1e-3)
|
||
|
||
|
||
def test_b1_max_envelope_exact_formula():
|
||
"""b1_model="max_envelope" (default): B1 = max(fraction_channels,
|
||
aggregate_bias) where
|
||
fraction_channels = C_B1 · g · W · log₂(1 + G/σ)
|
||
aggregate_bias = C_B1 · W · log₂(1 + g·G/σ)
|
||
Hand-compute for one config; verify the larger term is selected."""
|
||
g, gnm, sigma, W = 0.1, 2.0, 0.5, 5000
|
||
snr_per_channel = gnm / sigma # 4.0
|
||
snr_grad = g * gnm / sigma # 0.4
|
||
fc = 1.0 * g * W * math.log2(snr_per_channel + 1) # 0.1·5000·log₂(5) ≈ 1160.96
|
||
ab = 1.0 * W * math.log2(snr_grad + 1) # 5000·log₂(1.4) ≈ 2426.26
|
||
expected_b1 = max(fc, ab)
|
||
expected_selected = "aggregate_bias" if ab >= fc else "fraction_channels"
|
||
r = t3_bound_bits(**{
|
||
**_BASELINE,
|
||
"gradient_fraction": g, "gradient_norm_max": gnm,
|
||
"gradient_noise_stddev": sigma, "window_length": W,
|
||
# b1_model defaults to max_envelope
|
||
})
|
||
assert r["b1_model"] == "max_envelope"
|
||
assert r["b1_selected"] == expected_selected
|
||
assert r["B1_contribution"] == pytest.approx(expected_b1, abs=1e-3)
|
||
assert r["B1_fraction_channels"] == pytest.approx(fc, abs=1e-3)
|
||
assert r["B1_aggregate_bias"] == pytest.approx(ab, abs=1e-3)
|
||
assert r["snr_per_channel"] == pytest.approx(snr_per_channel, abs=1e-3)
|
||
|
||
|
||
@pytest.mark.parametrize("bad", ["v2", "envelope", "", "MAX_ENVELOPE", None, 1])
|
||
def test_invalid_b1_model_rejected(bad):
|
||
with pytest.raises(ValueError, match="b1_model"):
|
||
t3_bound_bits(**{**_BASELINE, "b1_model": bad})
|
||
|
||
|
||
# --- integration with §6 framework -----------------------------------
|
||
|
||
|
||
def test_total_equals_sum_of_three_contributions():
|
||
"""Closure check: I_window ≡ B1 + B2 + B3 (no missing term, no
|
||
double-counting). Multiple inputs to widen the cone."""
|
||
for window_length, lr_decision_interval in [
|
||
(100, 10),
|
||
(5000, 250),
|
||
(10000, 100),
|
||
(50000, 1000),
|
||
]:
|
||
args = {**_BASELINE, "window_length": window_length,
|
||
"lr_decision_interval": lr_decision_interval}
|
||
r = t3_bound_bits(**args)
|
||
total = r["B1_contribution"] + r["B2_contribution"] + r["B3_contribution"]
|
||
# Allow small rounding (round to 4 dp on each contribution).
|
||
assert r["I_window_bits_upper_bound"] == pytest.approx(total, abs=1e-3)
|
||
|
||
|
||
# --- B3 hand-computed formula (gap caught by the docs/calculator-test-
|
||
# patterns.md checklist, 2026-05-10) ---------------------------------
|
||
|
||
|
||
def test_b3_exact_formula():
|
||
"""B3 = C_B3 · ⌈W/E⌉ · log₂(N_b · σ_grad / ‖∇L_max‖) / 2 per
|
||
#000036 §5 (Bottou-Bousquet refinement).
|
||
|
||
Pairs the existing test_b1_exact_formula + test_b2_exact_formula
|
||
by hand-computing B3 from the spec formula. Closes the
|
||
calculator-test-patterns.md §3 checklist gap (item 3:
|
||
hand-formula assertions per output field).
|
||
"""
|
||
# Pick a config where the B3 factor (N_b · σ / ‖∇L_max‖) > 1
|
||
# so the floor doesn't kick in — that case is covered separately
|
||
# by test_b3_floor_at_zero_when_factor_below_one.
|
||
g, gnm, sigma, W = 0.05, 1.0, 0.5, 4096
|
||
K, R = 100, 8
|
||
N_b, E = 512, 512
|
||
factor = N_b * sigma / gnm # 256.0 > 1
|
||
epochs = -(-W // E) # ceil
|
||
expected_b3 = 1.0 * epochs * math.log2(factor) / 2.0
|
||
r = t3_bound_bits(
|
||
gradient_fraction=g, gradient_norm_max=gnm,
|
||
gradient_noise_stddev=sigma, lr_decision_interval=K,
|
||
lr_grid_size=R, window_length=W,
|
||
batches_per_epoch=N_b, steps_per_epoch=E,
|
||
)
|
||
assert r["B3_contribution"] == pytest.approx(expected_b3, abs=1e-3), (
|
||
f"B3 mismatch: function returned {r['B3_contribution']}, "
|
||
f"hand-computed {expected_b3} via "
|
||
f"⌈{W}/{E}⌉ · log₂({factor})/2 = {epochs} · {math.log2(factor):.4f}/2"
|
||
)
|
||
|
||
|
||
# --- KAT regression (gap caught by checklist, 2026-05-10) -----------
|
||
|
||
|
||
KAT_FIXTURE = (
|
||
pathlib.Path(__file__).parent.parent
|
||
/ "bench" / "fixtures" / "t3-bound"
|
||
/ "known-answer-tests.jsonl"
|
||
)
|
||
|
||
|
||
@pytest.mark.skipif(
|
||
not KAT_FIXTURE.exists(),
|
||
reason="KAT fixture not yet generated; run scripts/generate_t3_bound_kat.py",
|
||
)
|
||
def test_t3_bound_known_answer_tests():
|
||
"""Replay the pinned t3-bound/known-answer-tests.jsonl fixture;
|
||
every entry's per-contribution + total must match what the
|
||
function produces today.
|
||
|
||
Algorithm change MUST bump ``CALCULATOR_VERSION`` and emit a new
|
||
fixture file under ``bench/fixtures/t3-bound/`` — old runs
|
||
replay against old data per the calculator-test-patterns.md §1
|
||
discipline.
|
||
|
||
Closes the calculator-test-patterns.md §1 checklist gap (no
|
||
KAT fixture for this calculator).
|
||
"""
|
||
# Numeric fields checked on every KAT entry.
|
||
_NUMERIC = [
|
||
("expected_total", "I_window_bits_upper_bound"),
|
||
("expected_b1", "B1_contribution"),
|
||
("expected_b2", "B2_contribution"),
|
||
("expected_b3", "B3_contribution"),
|
||
("expected_snr_grad", "snr_grad"),
|
||
]
|
||
# Optional numeric fields (present in 2026-05-11+ fixtures).
|
||
_NUMERIC_OPT = [
|
||
("expected_b1_fraction_channels", "B1_fraction_channels"),
|
||
("expected_b1_aggregate_bias", "B1_aggregate_bias"),
|
||
("expected_b1_effective_control_v1", "B1_effective_control_v1"),
|
||
("expected_snr_per_channel", "snr_per_channel"),
|
||
]
|
||
n_kats = 0
|
||
for line in KAT_FIXTURE.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
kat = json.loads(line)
|
||
n_kats += 1
|
||
r = t3_bound_bits(**kat["inputs"])
|
||
assert r["calculator_version"] == kat["calculator_version"], (
|
||
f"KAT version mismatch on {kat['label']!r}: "
|
||
f"recorded {kat['calculator_version']}, "
|
||
f"function {r['calculator_version']}"
|
||
)
|
||
# b1_model + b1_selected pinned when recorded.
|
||
if "b1_model" in kat:
|
||
assert r["b1_model"] == kat["b1_model"], (
|
||
f"KAT b1_model mismatch on {kat['label']!r}: "
|
||
f"recorded {kat['b1_model']}, function {r['b1_model']}"
|
||
)
|
||
if "expected_b1_selected" in kat:
|
||
assert r["b1_selected"] == kat["expected_b1_selected"], (
|
||
f"KAT b1_selected mismatch on {kat['label']!r}: "
|
||
f"recorded {kat['expected_b1_selected']}, "
|
||
f"function {r['b1_selected']}"
|
||
)
|
||
if "expected_certification_status" in kat:
|
||
assert r["certification_status"] == kat["expected_certification_status"], (
|
||
f"KAT certification_status mismatch on {kat['label']!r}: "
|
||
f"recorded {kat['expected_certification_status']}, "
|
||
f"function {r['certification_status']}"
|
||
)
|
||
for field, key in _NUMERIC + _NUMERIC_OPT:
|
||
if field not in kat:
|
||
continue
|
||
recorded = float(kat[field])
|
||
observed = float(r[key])
|
||
assert observed == pytest.approx(recorded, abs=1e-3), (
|
||
f"KAT {field} mismatch on {kat['label']!r}: "
|
||
f"recorded {recorded}, observed {observed}"
|
||
)
|
||
assert n_kats >= 5, (
|
||
f"KAT fixture seems incomplete: only {n_kats} entries; "
|
||
"calculator-test-patterns.md §1 wants ≥ 5"
|
||
)
|