| """Round5 whole-HOLD admission, registered historical budget and forced output checks.""" |
| from __future__ import annotations |
|
|
| import json |
| import threading |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from offload.launch.windows import pf1_window as pf |
| from offload.launch.windows import pp4_budget as b |
| from offload.launch.windows import pp4_cells as c |
| from offload.launch.windows import pp4_metrics as m |
| from offload.launch.windows import pp4_prerequisites as prerequisites |
| from offload.launch.windows import pp4_window as w |
| from offload.launch.windows.tests import test_pf1_window as base |
| from offload.launch.windows.tests.test_pp4_window import WORKER |
| from offload.prefill_maxlen.evidence import Json, number, record |
|
|
|
|
| def test_budget_recomputes_gap_scaling_worker_cuts_and_explicit_profile_retirement() -> None: |
| result = b.estimate(pair={'seconds': 600}) |
| assert result['total_s'] == 4743 and result['fits'] is True |
| assert result['headline_per_start_s'] == 220 |
| assert result['cells'] == ['short-b1', '32k-b1'] and result['all_cells_total_s'] == 5373 |
| assert result['profile_per_start_s'] == 0 |
| assert result['block_budget_s'] == {'s1-A': 1167., 's2-B': 1089., 's3-A': 1047.} |
| assert 'measured_complete_block_budget' in prerequisites.RETIRED_GATES |
| assert 'measured_complete_block_budget' not in prerequisites.GATES |
|
|
|
|
| def test_all_nine_measured_cold_cell_source_hashes_and_numeric_values() -> None: |
| plan = record(json.loads(b.PATH.read_text())) |
| paths: set[str] = set() |
| for value in record(plan['cells']).values(): |
| row = record(value) |
| for observation in record_list(row['observations']): |
| path = pf.CAMPAIGN/str(observation['source']) |
| assert pf.digest(path) == observation['sha256'] |
| paths.add(str(path)) |
| streams = record_list(record(json.loads(path.read_text()))['streams']) |
| assert all(record(s['usage'])['completion_tokens'] == 1500 for s in streams) |
| assert observation['max_ttft_s'] == max(number(s['ttft_s']) for s in streams) |
| spans = [max(number(x) for x in record(s['last_by_kind_s']).values()) |
| -min(number(x) for x in record(s['first_by_kind_s']).values()) for s in streams] |
| assert observation['max_decode_span_s'] == max(spans) |
| assert len(paths) == 9 |
|
|
|
|
| def record_list(value: Json) -> list[dict[str, Json]]: |
| assert isinstance(value, list) |
| return [record(v) for v in value] |
|
|
|
|
| def test_missing_admission_cannot_be_replaced_by_old_pair_pointer(tmp_path: Path) -> None: |
| path = tmp_path/'missing-admission.json' |
| with pytest.raises(FileNotFoundError): b.pair_reserve(path, WORKER.commit) |
| path.write_text('{"schema":"pp4-physical-pair-v1","complete":true}') |
| with pytest.raises(ValueError): b.pair_reserve(path, WORKER.commit) |
|
|
|
|
| def test_late_start_needs_all_remaining_blocks_and_cannot_skip_closing_off(tmp_path: Path) -> None: |
| clock = base.Clock() |
| live = w.Live(tmp_path, pf.Supervisor(tmp_path, clock=clock), native=WORKER, remote='experiment', |
| expected=WORKER.commit, data_dir=tmp_path, tag='owned', admission={ |
| 'verifier_sha256': {}, 'block_budget_s': {'s1-A': 100., 's2-B': 110., 's3-A': 100.}}, clock=clock) |
| try: |
| live.submit = clock()+310 |
| assert live.fits(w.ARMS[0]) |
| clock.now += .001 |
| assert not live.fits(w.ARMS[0]) |
| live.submit = clock()+210 |
| assert live.fits(w.ARMS[1]) |
| clock.now += .001 |
| assert not live.fits(w.ARMS[1]) |
| live.submit = clock()+100 |
| assert live.fits(w.ARMS[2]) |
| clock.now += .001 |
| assert not live.fits(w.ARMS[2]) |
| assert w.SUBMIT_S == w.ABORT_S == 73*60 and w.RESTORE_S == 85*60 |
| finally: |
| live.sup.close() |
|
|
|
|
| def test_seeded_http_preserves_settings_and_forces_eos() -> None: |
| seen: list[dict[str, Json]] = [] |
| class Api(c.Http): |
| def stream(self, body: dict[str, Json], first: threading.Event) -> dict[str, Json]: |
| seen.append(body) |
| return {} |
| c.SeededApi(Api('http://unused'), 45).stream({'seed': 42, 'max_tokens': 512, 'temperature': 1, 'top_p': 1, |
| 'cache_salt': 'owned-start-wave:cell:0'}, threading.Event()) |
| assert seen == [{'seed': 45, 'max_tokens': 512, 'temperature': 1, 'top_p': 1, 'ignore_eos': True, |
| 'cache_salt': 'owned-start-wave:cell:0'}] |
|
|
|
|
| @pytest.mark.parametrize('tokens,finish', [(511, 'length'), (1500, 'length'), (512, 'stop')]) |
| def test_short_eos_or_legacy_output_never_enters512_summary(tokens: int, finish: str) -> None: |
| with pytest.raises(ValueError, match='budget incomplete'): |
| c.summarize({'batch': 1, 'streams': [{'usage': {'completion_tokens': tokens}, 'finish': finish}]}, {}, {}) |
|
|
|
|
| def test_split_window_input_is_retired_not_reinterpreted() -> None: |
| assert [a.role for a in w.ARMS] == ['A', 'B', 'A'] |
| with pytest.raises(ValueError, match='one complete'): |
| w.validate_prior({}, 'a'*40) |
| assert m.REGISTRATION == 'pp4-r6-512-aba-no-measurement-profiles-v1' |
| assert [m.group_count(name) for name in c.CELLS] == [4, 4, 2] |
|
|
|
|
| def constant_groups(rate: int, count: int = 4) -> list[dict[str, Json]]: |
| return [{'spec_request_rounds': 100., 'accepted_tokens': float(rate*2-100), 'server_decode_seconds': 2.} |
| for _ in range(count)] |
|
|
|
|
| def test_strict_win_boundary_and_inclusive_loss_boundary() -> None: |
| equal = m.compare([constant_groups(100), constant_groups(102), constant_groups(100)]) |
| assert equal['required_delta_tps'] == equal['point_delta_tps'] == 2. |
| assert equal['win'] is False |
| assert m.compare([constant_groups(100), constant_groups(103), constant_groups(100)])['win'] is True |
| loss = m.compare([constant_groups(100), constant_groups(95), constant_groups(110)]) |
| assert loss['point_delta_tps'] == -10. and loss['off_spread_tps'] == 10. |
| assert loss['no_loss_beyond_spread'] is True |
| assert m.compare([constant_groups(100), constant_groups(94), constant_groups(110)])['no_loss_beyond_spread'] is False |
|
|
|
|
| def test_off_spread_overrides_two_percent_and_every_group_stays_intact() -> None: |
| point = m.compare([constant_groups(100, 2), constant_groups(115, 2), constant_groups(110, 2)]) |
| assert point['required_delta_tps'] == 10. and point['win'] is False |
| assert point['bootstrap_seed'] == 4103 and point['resamples'] == 2000 |
| assert point['independent_groups'] == 2 |
| assert point['paired_delta_95pct_tps'] == [10., 10.] |
|
|
|
|
| def test_b4_cannot_win_and_any_cell_loss_stops() -> None: |
| from offload.launch.windows.tests.test_pp4_window import complete |
| starts = complete() |
| on = record(record(starts[w.ARMS[1].name])['cells']) |
| for name in ('short-b1', '32k-b1'): |
| for row in record_list(record(on[name])['waves']): row['accepted_tokens'] = 1000. |
| assert w.decision(starts)['status'] == 'INCONCLUSIVE' |
| starts = complete() |
| for row in record_list(record(record(record(starts[w.ARMS[1].name])['cells'])['32k-b4'])['waves']): |
| row['accepted_tokens'] = 999. |
| assert w.decision(starts)['status'] == 'INCONCLUSIVE' |
|
|
|
|
| @pytest.mark.parametrize('change', ['512', 'eos', 'seed', 'salt', 'complete', 'greedy']) |
| def test_each_evidence_contract_refuses_changed_receipt(change: str) -> None: |
| from offload.launch.windows.tests.test_pp4_window import complete |
| starts = complete() |
| block = record(starts[w.ARMS[1].name]) |
| group = record_list(record(record(block['cells'])['short-b1'])['waves'])[0] |
| stream = record_list(group['streams'])[0] |
| if change == '512': record(stream['usage'])['completion_tokens'] = 511 |
| if change == 'eos': record(stream['sampling'])['ignore_eos'] = False |
| if change == 'seed': record(stream['sampling'])['seed'] = 43 |
| if change == 'salt': |
| off = record_list(record(record(record(starts[w.ARMS[0].name])['cells'])['short-b1'])['waves'])[0] |
| stream['cache_salt_sha256'] = record_list(off['streams'])[0]['cache_salt_sha256'] |
| if change == 'complete': group['completed_requests'] = 0 |
| if change == 'greedy': record_list(record(block['greedy'])['checks'])[5]['identical'] = False |
| with pytest.raises(ValueError): w.decision(starts) |
|
|
|
|
| def test_nearest_rank95_interval_is_preregistered() -> None: |
| assert m.interval(list(range(2000))) == [49, 1949] |
|
|
|
|
| def test_actual_execute_refuses_budget_before_output_or_process(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, |
| capsys: pytest.CaptureFixture[str]) -> None: |
| import subprocess |
| monkeypatch.setenv('WINDOW_COORDINATOR', 'Deepseek/coordinator') |
| monkeypatch.setattr(b, 'pair_reserve', lambda path, expected: {'seconds': 960}) |
| proof = tmp_path/'primary.json'; proof.write_text('{}') |
| receipt: dict[str, Json] = {'schema': 'pp4-live-admission-v1', 'source_commit': WORKER.commit, |
| 'verdict': 'QUALIFIED', 'depth': 5, 'worker_identity': WORKER.identity, 'markov_identity': WORKER.markov, |
| 'runtime_python': prerequisites.PYTHON, 'runtime_torch': '2.12.1+cu130', |
| 'runtime_artifact_sha256': {prerequisites.STOCK: 'd'*64, prerequisites.FACC: 'e'*64, |
| 'p2p/'+pf.digest(pf.ROOT/'offload/dspark/worker/p2p/p2p.cu')[:16]+'/dspark_p2p.so': 'f'*64}, |
| 'gates': {name: {'pass': True, 'evidence_path': str(proof.resolve()), 'sha256': pf.digest(proof)} |
| for name in prerequisites.GATES}} |
| path = tmp_path/'qualification.json'; path.write_text(json.dumps(receipt)) |
| def forbidden(*args: object, **kwargs: object) -> None: raise AssertionError('process/remote action before budget refusal') |
| monkeypatch.setattr(subprocess, 'Popen', forbidden) |
| out = tmp_path/'PP4-never-created' |
| assert w.main(['--execute', '--qualification', str(path), '--expected-experiment-commit', WORKER.commit, |
| '--out', str(out)]) == 1 |
| assert not out.exists() |
| assert 'ESTIMATED 85.05min exceeds85min after B4 omission' in capsys.readouterr().err |
|
|
