File size: 8,726 Bytes
1e251f3 | 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 | """WIN2-R3 process rules as code: two-window cap and 2x resolution filter (CPU only, tmp campaign and archive)."""
from __future__ import annotations
import json
import math
from pathlib import Path
import pytest
from offload.launch.windows import decision_v2 as d
from offload.launch.windows import resolution as rs
from offload.launch.windows import window_cap as cap
from offload.prefill_maxlen.evidence import Json, number, record
PROD = '596ba1cf4e3b11be29b41a57e7f69a5d8efa4799'
SINCE = cap.utc('2026-10-27T00:00:00Z')
def _window(camp: Path, name: str, status: str, **summary: Json) -> Path:
w = camp / name
w.mkdir(parents=True)
(w / 'summary.json').write_text(json.dumps({'status': status, 'decision': {'status': 'NOT A CANDIDATE'}, **summary}))
return w
def test_only_completed_measurement_windows_count(tmp_path: Path) -> None:
_window(tmp_path, 'PC1-a-W-20261027T100000Z', 'COMPLETE')
_window(tmp_path, 'PC1-b-W-20261027T110000Z', 'INCOMPLETE') # stopped: never counts
_window(tmp_path, 'PC1-admission-W-20261027T120000Z', 'COMPLETE')
_window(tmp_path, 'PC1-c-W-20261027T130000Z', 'COMPLETE', kind='smoke')
_window(tmp_path, 'PC1-d-W-20261026T130000Z', 'COMPLETE') # before adoption
_window(tmp_path, 'UP3-e-W-20261027T140000Z', 'COMPLETE') # another lane
r = cap.check('PC1', since=SINCE, campaign=tmp_path)
counted = r['counted']
assert isinstance(counted, list)
assert r['verdict'] == 'ALLOW' and [record(w)['window'] for w in counted] == ['PC1-a-W-20261027T100000Z']
def test_third_window_needs_a_hypothesis_dated_after_the_second(tmp_path: Path) -> None:
_window(tmp_path, 'PC1-a-W-20261027T100000Z', 'COMPLETE')
_window(tmp_path, 'PC1-b-W-20261027T150000Z', 'INCONCLUSIVE')
r = cap.check('PC1', since=SINCE, campaign=tmp_path)
assert r['verdict'] == 'REFUSE' and 'hypothesis-2.md' in str(r['reason'])
hyp = tmp_path / 'experiments/PC1/hypothesis-2.md'
hyp.parent.mkdir(parents=True)
hyp.write_text('# PC1 hypothesis 2\ndate: 2026-10-27T14:00Z\n') # dated BEFORE the second window
assert cap.check('PC1', since=SINCE, campaign=tmp_path)['verdict'] == 'REFUSE'
hyp.write_text('# PC1 hypothesis 2\ndate: 2026-10-27T16:30Z\nchunk 8192 changes ...\n')
assert cap.check('PC1', since=SINCE, campaign=tmp_path)['verdict'] == 'ALLOW'
_window(tmp_path, 'PC1-c-W-20261027T170000Z', 'COMPLETE') # the 4th needs another one
assert cap.check('PC1', since=SINCE, campaign=tmp_path)['verdict'] == 'REFUSE'
@pytest.mark.parametrize('line,verdict', [('| HYPOTHESIS PC1 3 | 2026-10-27 18:10 | depth 1 |', 'ALLOW'),
('2026-10-27T18:00Z WINDOW-CAP OVERRIDE PC1 (NDM: one more)', 'ALLOW'),
('2026-10-27T18:00Z HYPOTHESIS PC10 3', 'REFUSE'),
('HYPOTHESIS PC1 3 (undated)', 'REFUSE')])
def test_register_line_or_override(tmp_path: Path, line: str, verdict: str) -> None:
_window(tmp_path, 'PC1-a-W-20261027T100000Z', 'COMPLETE')
_window(tmp_path, 'PC1-b-W-20261027T150000Z', 'COMPLETE')
(tmp_path / 'LANES.md').write_text('# register\n' + line + '\n')
assert cap.check('PC1', since=SINCE, campaign=tmp_path)['verdict'] == verdict
def test_cli_needs_adoption_and_prefixes(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
_window(tmp_path, 'NBW2-W-20261027T100000Z', 'COMPLETE')
_window(tmp_path, 'NBW-W-20261027T110000Z', 'COMPLETE')
assert cap.main(['NBW', '--campaign', str(tmp_path)]) == 2 # not adopted: refuses
(tmp_path / cap.ADOPTION_FILE).write_text(json.dumps({'adopted_utc': '2026-10-27T00:00:00Z'}))
assert cap.main(['NBW', '--campaign', str(tmp_path), '--prefix', 'NBW-W', '--prefix', 'NBW2-W']) == 2
capsys.readouterr()
assert cap.main(['NBW', '--campaign', str(tmp_path), '--prefix', 'NBW2-W']) == 0
assert json.loads(capsys.readouterr().out)['verdict'] == 'ALLOW'
with pytest.raises(ValueError):
cap.check('../x', since=SINCE, campaign=tmp_path)
# ---------------------------------------------------------------- 2x resolution filter
def _archive(tmp_path: Path, cv: float, windows: int = 3, n: int = 6, metric: str = 'ms_per_round',
prompts: tuple[str, ...] = ('0' * 64,)) -> Path:
path = tmp_path / 'a.jsonl'
rows: list[dict[str, Json]] = []
for w in range(windows):
base = 40.0 * (1 + 0.01 * w)
# n values with exactly sample SD = cv * mean: alternate +-k around the mean
k = cv * base * math.sqrt((n - 1) / n)
for i in range(n):
rows.append({'window': f'W{w}', 'cell': '32k-b1', 'metric': metric, 'value': base + (k if i % 2 else -k),
'max_tokens': 512, 'commit': PROD, 'clock_cap_mhz': 2150, 'date': '20261027',
'start': 's1-control', 'repeat': i, 'prompt_sha256': prompts[w % len(prompts)]})
d.archive_append(path, rows)
return path
def test_resolution_never_pools_across_prompt_hashes(tmp_path: Path) -> None:
"""DP-2: a definition whose archive rows come from two prompts refuses unless one prompt is named."""
path = _archive(tmp_path, .02, windows=4, prompts=('0' * 64, '1' * 64))
with pytest.raises(ValueError, match='never pooled across prompt hashes'):
rs.pooled_cv(path, '32k-b1', 'ms_per_round', max_tokens=512, clock_cap_mhz=2150, commit=PROD)
one = rs.pooled_cv(path, '32k-b1', 'ms_per_round', max_tokens=512, clock_cap_mhz=2150, commit=PROD,
prompt_sha256='1' * 64)
assert one['windows'] == 2 and one['prompt_sha256'] == '1' * 64
args = ['--cell', '32k-b1', '--metric', 'ms_per_round', '--design', 'unpaired:4v3', '--expected-gain', '0.5',
'--max-tokens', '512', '--clock', '2150', '--commit', PROD, '--archive', str(path)]
assert rs.main(args) == 2 and rs.main([*args, '--prompt-sha256', '0' * 64]) == 0
def test_pooled_cv_and_half_widths_match_the_formulas(tmp_path: Path) -> None:
path = _archive(tmp_path, .02)
got = rs.pooled_cv(path, '32k-b1', 'ms_per_round', max_tokens=512, clock_cap_mhz=2150, commit=PROD)
assert number(got['cv']) == pytest.approx(.02) and got['windows'] == 3 and got['df'] == 15
assert number(rs.half_width('unpaired:3v3', .021)['half_width']) == pytest.approx(2.776 * .021 * math.sqrt(2 / 3))
assert number(rs.half_width('pairs:6', .021)['half_width']) == pytest.approx(2.571 * .021 * math.sqrt(2) / math.sqrt(6))
est = rs.half_width('seeded:6', 0.0, ms_cv=.01)
assert number(est['half_width']) == pytest.approx(2.571 * .01 * math.sqrt(2) / math.sqrt(6)) and 'ESTIMATED' in str(est['basis'])
measured = rs.half_width('seeded:12', 0.0, diff_cv=.02)
assert number(measured['half_width']) == pytest.approx(2.201 * .02 / math.sqrt(12)) and 'MEASURED' in str(measured['basis'])
for bad in ('3v3', 'unpaired:1v3', 'seeded:4', 'pairs:x'):
with pytest.raises(ValueError):
rs.half_width(bad, .02, ms_cv=.01)
def test_filter_refuses_below_twice_the_half_width(tmp_path: Path) -> None:
path = _archive(tmp_path, .01)
def verdict(metric: str, design: str, gain: float) -> dict[str, Json]:
return rs.check('32k-b1', metric, design, gain, max_tokens=512, clock_cap_mhz=2150, commit=PROD, archive=path)
seeded6 = verdict('decode_tps', 'seeded:6', .03)
assert seeded6['verdict'] == 'QUEUE' and number(seeded6['required_gain']) < .03 # +-1.48 % (F-METHOD)
assert verdict('ms_per_round', 'unpaired:3v3', .05)['verdict'] == 'QUEUE'
assert verdict('ms_per_round', 'unpaired:3v3', .04)['verdict'] == 'REFUSE' # +-2.27 %
with pytest.raises(ValueError, match='no archive history'):
rs.check('32k-b1', 'ms_per_round', 'unpaired:3v3', .03, max_tokens=512, clock_cap_mhz=2300, commit=PROD,
archive=path)
def test_cli_exit_codes(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
path = _archive(tmp_path, .01)
base = ['--cell', '32k-b1', '--metric', 'decode_tps', '--max-tokens', '512', '--clock', '2150', '--commit', PROD,
'--archive', str(path)]
assert rs.main([*base, '--design', 'seeded:6', '--expected-gain', '0.03']) == 0
assert record(json.loads(capsys.readouterr().out))['verdict'] == 'QUEUE'
assert rs.main([*base, '--design', 'seeded:6', '--expected-gain', '0.01']) == 2
assert rs.main([*base, '--design', 'seeded:6', '--expected-gain', '0.03', '--diff-cv', '0.05']) == 2
assert rs.main([*base, '--design', 'seeded:3', '--expected-gain', '0.03']) == 2
|