"""CPU checks of R-PP4's WIN2 measurement: win2-v1 decisions, production-policy arm, carried qualification, chain mode. All serving, host and GPU operations are fakes; the only real files read are the registered round6 budget, the frozen production env at PROD_COMMIT and (when present) a real PP4 window's rollback.json. """ from __future__ import annotations import copy import json import os import shlex import signal import subprocess from pathlib import Path from typing import cast import pytest from offload.launch.windows import chain from offload.launch.windows import decision_v2 as d from offload.launch.windows import pf1_window as pf from offload.launch.windows import pp4_budget as budget from offload.launch.windows import pp4_cells as cells from offload.launch.windows import pp4_measure_v2 as m from offload.launch.windows import pp4_prerequisites as prereq from offload.launch.windows import pp4_window as w from offload.launch.windows import scoped_remote as owner from offload.launch.windows.tests import test_pf1_window as base from offload.launch.windows.tests import test_pp4_window as prior from offload.prefill_maxlen.evidence import Json, number, record GATE: dict[str, Json] = {'status': 'PASS', 'receipt': '/fixture', 'pp4_kernel_sha256': 'a' * 64} CLOCK_OK: dict[str, Json] = {'consistent': True, 'reason': None} REAL_ROLLBACK = pf.CAMPAIGN / 'PP4-429-admission-W-20261027T133630Z/rollback.json' def v2_cell(name: str, start: str, *, accepted: float = 200., seconds: float = 4., tps: float = 100.) -> dict[str, Json]: """Six registered waves; repeat r adds content variation shared by every start (paired by repeat).""" rows: list[Json] = [] for wave in range(m.WAVES): r = wave // m.WAVES_PER_REPEAT rows.append({'wave': wave, 'repeat': r, 'family': cells.FAMILIES[wave % 2], 'batch': 1, 'spec_request_rounds': 100., 'accepted_tokens': float(round(accepted * (1 + .05 * r) + wave % 2)), 'server_decode_seconds': seconds * (1 + .01 * r), 'median_ttft_s': 1. + .1 * r, 'median_decode_tps': tps * (1 + .02 * r), 'completed_requests': 1, 'streams': [{'prompt_sha256': f'{name}-{m.ORDERS[name][wave]}-{cells.FAMILIES[wave % 2]}', 'cache_salt_sha256': f'{start}-{name}-{wave}', 'sampling': {'temperature': 1, 'top_p': 1, 'seed': 42 + wave, 'priority': 10, 'max_tokens': 512, 'ignore_eos': True}, 'usage': {'completion_tokens': 512}, 'finish': 'length'}]}) return {'cell': name, 'batch': 1, 'metric_registration': m.REGISTRATION, 'waves': rows} GREEDY: dict[str, Json] = {'greedy_registration': cells.GREEDY_REGISTRATION, 'identical': 6, 'near_tie': 0, 'checks': [{'i': i, 'identical': True} for i in range(6)]} def starts(mode: str, *, candidate: dict[str, float] | None = None) -> dict[str, Json]: out: dict[str, Json] = {} for name, role in m.MODES[mode]: values = candidate if role == 'B' and candidate is not None else {} block: dict[str, Json] = {'role': role, 'status': 'COMPLETE', 'cells': {c: v2_cell(c, name, **values) for c in m.CELLS}} if name in m.GREEDY[mode]: block['greedy'] = copy.deepcopy(GREEDY) out[name] = block return out WIN = {'accepted': 230., 'seconds': 4.1, 'tps': 108.} # +15 % accepted, +2.5 % ms/round: rate +7 % def test_v2_prompts_are_distinct_per_family_and_cuts_exist() -> None: for name in m.CELLS: for family in cells.FAMILIES: keys = [m.wave_prompts(name, w_)[0] for w_ in range(m.WAVES) if cells.FAMILIES[w_ % 2] == family] assert len(set(keys)) == m.REPEATS assert all(m.wave_prompts(name, w_)[0] in cells.PROMPTS for w_ in range(m.WAVES)) with pytest.raises(ValueError): m.wave_prompts('32k-b4', 0) def test_repeat_values_pool_exact_counters() -> None: rows = m.waves(v2_cell('short-b1', 's1-A')) reps = m.repeat_values(rows) assert [r['waves'] for r in reps] == [[0, 1], [2, 3], [4, 5]] first = reps[0] assert first['accepted_per_round'] == (200. + 201.) / 200 assert first['ms_per_round'] == 1000 * 8. / 200 assert first[m.PRIMARY] == pytest.approx((200 + 401) / 8.) @pytest.mark.parametrize('change', ['registration', 'order', 'seed', 'duplicate-prompt', 'duplicate-salt', 'short-output']) def test_wave_evidence_refusals(change: str) -> None: cell = v2_cell('32k-b1', 's1-A') rows = cast(list[dict[str, Json]], cell['waves']) stream = record(cast(list[Json], rows[3]['streams'])[0]) if change == 'registration': cell['metric_registration'] = 'pp4-r6-512-aba-no-measurement-profiles-v1' elif change == 'order': rows[2]['family'], rows[3]['family'] = rows[3]['family'], rows[2]['family'] elif change == 'seed': record(stream['sampling'])['seed'] = 42 elif change == 'duplicate-prompt': stream['prompt_sha256'] = record(cast(list[Json], rows[1]['streams'])[0])['prompt_sha256'] elif change == 'duplicate-salt': stream['cache_salt_sha256'] = record(cast(list[Json], rows[1]['streams'])[0])['cache_salt_sha256'] else: record(stream['usage'])['completion_tokens'] = 511 with pytest.raises(ValueError): m.waves(cell) def test_controlled_win_is_decided_on_the_rate_not_on_ms_per_round() -> None: result = m.decide(starts('controlled', candidate=WIN), 'controlled', clock=CLOCK_OK, rng=GATE, archive=None) assert result['status'] == 'CANDIDATE' and result['rule'] == d.RULE == 'win2-v1' and result['min_gain'] == .02 assert record(result['groups'])['single-user'] == 'CANDIDATE' for cell in m.CELLS: metrics = record(record(record(result['cells'])[cell])['metrics']) assert list(metrics) == [m.PRIMARY, 'accepted_per_round', 'decode_tps'] assert record(metrics[m.PRIMARY])['verdict'] == 'WIN' reported = record(record(result['reported_metrics'])[cell]) assert number(record(reported['ms_per_round'])['point_gain']) < 0 # the cost PP4 pays, visible and reported # What WIN2.md's ms/round-primary default would have said about the same evidence. assert record(result['checklist_literal'])['status'] != 'CANDIDATE' def recompute(series_rows: Json) -> dict[str, Json]: series = [d.Series(str(r['cell']), str(r['metric']), bool(r['higher_is_better']), tuple(number(v) for v in cast(list[Json], r['control'])), tuple(number(v) for v in cast(list[Json], r['candidate'])), bool(r['paired'])) for r in (record(x) for x in cast(list[Json], series_rows))] return d.decide_window(series, m.CELLS) @pytest.mark.parametrize('candidate,clock', [(WIN, CLOCK_OK), ({'accepted': 150.}, CLOCK_OK), (WIN, {'consistent': False, 'reason': 'caps differ'})]) def test_decision_recomputes_from_its_recorded_series(candidate: dict[str, float], clock: dict[str, Json]) -> None: """The consumer contract (pp4_promote measurement_gate): rule/status/min_gain/cells recompute from decision_series.""" result = m.decide(starts('production-policy', candidate=candidate), 'production-policy', clock=clock, rng=GATE, archive=None) again = recompute(result['decision_series']) same = all(json.dumps(again[k], sort_keys=True) == json.dumps(result[k], sort_keys=True) for k in ('rule', 'status', 'min_gain', 'decision_cells', 'cells')) assert same is bool(clock.get('consistent')) # a clock-guarded verdict must NOT pass as recomputed evidence rows = [record(r) for r in cast(list[Json], result['decision_series'])] assert [(r['cell'], r['metric']) for r in rows] == [(c, k) for c in m.CELLS for k in (m.PRIMARY, 'accepted_per_round', 'decode_tps')] assert {(record(r)['cell'], record(r)['metric']) for r in cast(list[Json], result['reported_series'])} == { (c, k) for c in m.CELLS for k in ('ms_per_round', 'ttft_s')} def test_controlled_loss_and_guard_loss() -> None: worse = m.decide(starts('controlled', candidate={'accepted': 150.}), 'controlled', clock=CLOCK_OK, rng=GATE, archive=None) assert worse['status'] == 'NOT A CANDIDATE' # Rate better through a much cheaper round, but client tok/s collapses: the guard alone makes it a LOSS. guard = m.decide(starts('controlled', candidate={'accepted': 200., 'seconds': 3.5, 'tps': 60.}), 'controlled', clock=CLOCK_OK, rng=GATE, archive=None) assert guard['status'] == 'NOT A CANDIDATE' metrics = record(record(record(guard['cells'])['short-b1'])['metrics']) assert record(metrics[m.PRIMARY])['verdict'] == 'WIN' and record(metrics['decode_tps'])['verdict'] == 'LOSS' def test_production_policy_mode_requires_greedy_on_first_p_and_b() -> None: rows = starts('production-policy', candidate=WIN) result = m.decide(rows, 'production-policy', clock=CLOCK_OK, rng=GATE, archive=None) assert result['status'] == 'CANDIDATE' and 'not the production binary' in str(result['comparison']) record(rows['s1-P']).pop('greedy') with pytest.raises(KeyError): m.decide(rows, 'production-policy', clock=CLOCK_OK, rng=GATE, archive=None) rows = starts('production-policy', candidate=WIN) record(cast(list[Json], record(record(rows['s1-P'])['greedy'])['checks'])[2])['identical'] = False with pytest.raises(ValueError, match='s1-P'): m.decide(rows, 'production-policy', clock=CLOCK_OK, rng=GATE, archive=None) @pytest.mark.parametrize('change', ['clock', 'incomplete', 'missing-start', 'rng', 'inputs', 'role']) def test_decision_refusals_and_inconclusive(change: str) -> None: rows = starts('controlled', candidate=WIN) clock: dict[str, Json] = dict(CLOCK_OK) rng: dict[str, Json] | None = GATE if change == 'clock': clock = {'consistent': False, 'reason': 'thermal watchdog dialled the clock back during the HOLD'} elif change == 'incomplete': record(rows['s3-A'])['status'] = 'STARTED' elif change == 'missing-start': rows.pop('s3-A') elif change == 'rng': rng = {'status': 'FAIL'} elif change == 'inputs': wave = record(cast(list[Json], record(record(record(rows['s2-B'])['cells'])['short-b1'])['waves'])[0]) record(cast(list[Json], wave['streams'])[0])['prompt_sha256'] = 'other' else: record(rows['s2-B'])['role'] = 'A' if change in ('rng', 'inputs', 'role'): with pytest.raises(ValueError): m.decide(rows, 'controlled', clock=clock, rng=rng, archive=None) else: assert m.decide(rows, 'controlled', clock=clock, rng=rng, archive=None)['status'] == 'INCONCLUSIVE' def test_clock_guard_and_thermal_log(tmp_path: Path) -> None: text = 'sparkhead clock cap 2300 MHz\nsparkworker clock cap 2300 MHz\nport 8001 restricted\n' caps = m.caps_from_start(text) assert caps == {'sparkhead': 2300, 'sparkworker': 2300} names = ['s1-A', 's2-B', 's3-A'] assert m.clock_guard({n: caps for n in names}, names, [])['consistent'] is True assert m.clock_guard({n: {} for n in names}, names, [])['consistent'] is True # stock clocks throughout assert m.clock_guard({**{n: caps for n in names}, 's3-A': {'sparkhead': 2150, 'sparkworker': 2150}}, names, [])['consistent'] is False assert m.clock_guard({'s1-A': caps, 's2-B': caps}, names, [])['consistent'] is False log = tmp_path / 'THERMAL.log' log.write_text('2026-10-27T20:00:00Z sparkhead 70, 2300, 100, Not Active, Not Active\n' '2026-10-27T20:10:00Z DIAL BACK to 2150: sparkhead 91, ...\n') import datetime at = datetime.datetime(2026, 10, 27, 20, 10, tzinfo=datetime.timezone.utc).timestamp() assert m.dial_backs(log, at - 60, at + 60) == [{'utc': '2026-10-27T20:10:00Z', 'mhz': 2150}] assert m.dial_backs(log, at + 1, at + 60) == [] assert m.clock_guard({n: caps for n in names}, names, m.dial_backs(log, at - 60, at + 60))['consistent'] is False def test_budget_v2_fits_the_hold_and_names_every_start() -> None: pair: dict[str, Json] = {'seconds': 333, 'classification': 'fixture'} controlled = m.budget_v2('controlled', pair=pair) assert controlled['block_budget_s'] == {'s1-A': 1277., 's2-B': 1199., 's3-A': 1157.} assert controlled['total_s'] == 4806. and controlled['measurement_end_s'] == 4086. and controlled['fits'] is True production = m.budget_v2('production-policy', pair=pair) assert production['block_budget_s'] == {'s1-P': 1319., 's2-B': 1199., 's3-P': 1157.} and production['fits'] is True assert m.budget_v2('controlled')['fits'] is True # 600 s dry bound: 84.55 min assert m.budget_v2('production-policy')['fits'] is False # 600 s dry bound: 85.25 min; refuses def fake_git(changed: list[str], *, qualified: str = '1' * 40, runner_old: str | None = None) -> object: current = (pf.ROOT / m.RUNNER).read_text() def git(args: list[str]) -> str: if args[0] == 'merge-base': return qualified + '\n' if args[0] == 'diff': return '\n'.join(changed) + '\n' if args[0] == 'show': return runner_old if runner_old is not None and args[1].startswith(qualified) else current raise AssertionError(args) return git @pytest.mark.parametrize('changed,ok', [ ([m.RUNNER, 'offload/launch/windows/pp4_measure_v2.py', 'offload/gb10_budget/window_cells.py', 'offload/launch/windows/tests/test_pp4_window.py', 'documents/plans/x.md'], True), (['offload/dspark/worker/server.py'], False), (['offload/launch/windows/PP4-sampled.env'], False), (['offload/dspark/spark/remote_dspark/speculator.py'], False), (['offload/launch/windows/pp4_admission.py'], False)]) def test_carried_qualification_allows_only_the_measurement_layer(monkeypatch: pytest.MonkeyPatch, changed: list[str], ok: bool) -> None: monkeypatch.setattr(prereq, 'qualification', lambda path, expected: {'verdict': 'QUALIFIED', 'source_commit': expected}) monkeypatch.setattr(budget, 'pair_reserve', lambda path, expected: {'seconds': 333, 'classification': 'fixture'}) git = fake_git(changed) if ok: data = m.carried(Path('/receipt.json'), '1' * 40, '2' * 40, 'controlled', git=git) # type: ignore[arg-type] assert record(data['registered_budget'])['fits'] is True assert data['block_budget_s'] == record(data['registered_budget'])['block_budget_s'] assert record(data['carried_qualification'])['qualified_commit'] == '1' * 40 else: with pytest.raises(ValueError, match='new admission'): m.carried(Path('/receipt.json'), '1' * 40, '2' * 40, 'controlled', git=git) # type: ignore[arg-type] def test_carried_qualification_refuses_runner_unit_drift(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(prereq, 'qualification', lambda path, expected: {'verdict': 'QUALIFIED'}) monkeypatch.setattr(budget, 'pair_reserve', lambda path, expected: {'seconds': 333, 'classification': 'fixture'}) current = (pf.ROOT / m.RUNNER).read_text() old = current.replace("'timeout --signal=TERM --kill-after=15s --foreground 120 bash ~/dsv41-offload/prod_window.sh pair COORD\\n'", "'bash ~/dsv41-offload/prod_window.sh pair COORD\\n'") assert old != current with pytest.raises(ValueError, match='worker_script'): m.carried(Path('/r.json'), '1' * 40, '2' * 40, 'controlled', git=fake_git([m.RUNNER], runner_old=old)) # type: ignore[arg-type] monkeypatch.setattr(budget, 'pair_reserve', lambda path, expected: {'seconds': 600, 'classification': 'fixture'}) with pytest.raises(ValueError, match='exceeds'): m.carried(Path('/r.json'), '1' * 40, '2' * 40, 'production-policy', git=fake_git([])) # type: ignore[arg-type] def test_archive_rows_carry_max_tokens_and_only_finished_control_starts(tmp_path: Path) -> None: rows = starts('controlled', candidate=WIN) clock: dict[str, Json] = {'per_start': {n: {'sparkhead': 2300, 'sparkworker': 2300} for n in ('s1-A', 's2-B', 's3-A')}} archive = tmp_path / 'archive.jsonl' result = w.append_archive({'starts': rows}, 'controlled', window_name='PP4-v2-W-20261027T210000Z', experiment_commit='2' * 40, clock=clock, path=archive) assert {k: result[k] for k in ('status', 'rows', 'path')} == {'status': 'APPENDED', 'rows': 60, 'path': str(archive)} lines = [json.loads(x) for x in archive.read_text().splitlines()] assert {x['start'] for x in lines} == {'s1-A', 's3-A'} and all(x['max_tokens'] == 512 for x in lines) assert all(x['clock_cap_mhz'] == 2300 and x['clock_source'] == 'start-log' for x in lines) assert result['archive_head_before'] == {'rows': 0, 'head_sha256': d.GENESIS} assert record(result['archive_head_after'])['rows'] == 60 assert {x['cell'] for x in lines} == {'pp4rt-fixed5-greedy-short-b1', 'pp4rt-fixed5-greedy-32k-b1'} assert all(x['clock_caps_mhz'] == [2300] and x['date'] == '20261027' and x['commit'] == w.PROD_COMMIT for x in lines) where: dict[str, Json] = {'max_tokens': 512, 'clock_cap_mhz': 2300, 'commit': w.PROD_COMMIT} assert d.archive_spread(archive, 'pp4rt-fixed5-greedy-short-b1', m.PRIMARY, where=where) is None record(rows['s3-A'])['status'] = 'STARTED' again = w.append_archive({'starts': rows}, 'controlled', window_name='PP4-v2-W-20261028T010000Z', experiment_commit='2' * 40, clock=clock, path=archive) assert again['rows'] == 30 assert d.archive_spread(archive, 'pp4rt-fixed5-greedy-short-b1', m.PRIMARY, where=where) == 0.0 stock: dict[str, Json] = {'per_start': {n: {} for n in ('s1-A', 's2-B', 's3-A')}} # no cap printed: refuses loudly assert w.append_archive({'starts': rows}, 'controlled', window_name='PP4-v2-W-20261028T020000Z', experiment_commit='2' * 40, clock=stock, path=archive)['status'] == 'FAILED' broken = tmp_path / 'broken.jsonl' broken.write_text('{"schema": "other"}\n') assert w.append_archive({'starts': rows}, 'controlled', window_name='x', experiment_commit='2' * 40, clock=clock, path=broken)['status'] == 'FAILED' # ---- runner integration: real window() with a fake backend class FakeV2(prior.Fake): def __init__(self, fail: str = '', *, down: bool = False, chain_fail: bool = False, candidate: dict[str, float] | None = None) -> None: super().__init__(fail) self.down, self.chain_fail = down, chain_fail self.candidate = WIN if candidate is None else candidate self.restored_with: bool | None = None def cell(self, arm: w.Arm, name: str) -> dict[str, Json]: self.events.append('cell-' + arm.name + '-' + name) if self.fail == 'cell': raise TimeoutError('CENSORED request') return v2_cell(name, arm.name, **(self.candidate if arm.role == 'B' else {})) def greedy(self, arm: w.Arm) -> dict[str, Json]: self.events.append('greedy-' + arm.name) return copy.deepcopy(GREEDY) def production_down(self) -> bool: return self.down def owns_hold(self) -> bool: return self.fail != 'hold' def chain_abort(self) -> dict[str, Json]: self.events.append('chain-abort') return {'status': 'RESTORED: chain ended by abort; watch PASS; marker cleared'} def measurement_context(self) -> dict[str, Json]: return {'clock': dict(CLOCK_OK), 'rng_gate': dict(GATE)} def chain_close(self) -> dict[str, Json]: self.events.append('chain-close') if self.chain_fail: raise ValueError('quiet proof failed') return {'status': w.CHAINED, 'restore_deadline_met': True} def restore(self, *, production_stopped: bool) -> dict[str, Json]: self.restored_with = production_stopped if not self.owns_hold() and production_stopped: # Live.restore: no owned HOLD while production is down self.events.append('restore') raise ValueError('PP4 production may be stopped without owned HOLD') return super().restore(production_stopped=production_stopped) @pytest.mark.parametrize('mode', list(m.MODES)) def test_v2_window_order_greedy_and_decision(mode: str, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv('WINDOW_CHAIN', raising=False) fake = FakeV2() arms = w.V2_ARMS[mode] result = w.window(fake, lambda row: None, selected=arms, measured_cells=m.CELLS, v2=mode) assert result['status'] == 'COMPLETE' and record(result['decision'])['status'] == 'CANDIDATE' assert 'decision_series' in result and 'decision_series' not in record(result['decision']) assert recompute(result['decision_series'])['status'] == 'CANDIDATE' assert [x for x in fake.events if x in [a.name for a in arms]] == [a.name for a in arms] assert [x for x in fake.events if x.startswith('greedy-')] == [f'greedy-{n}' for n in m.GREEDY[mode]] assert fake.events[-1] == 'restore' and 'chain' not in result assert result['chain_state_at_start'] == 'production serving' if mode == 'production-policy': assert [a.config for a in arms] == [w.PRODUCTION_CONFIG, w.CONFIG, w.PRODUCTION_CONFIG] def test_v2_window_refuses_unregistered_cells_or_arms() -> None: fake = FakeV2() result = w.window(fake, lambda row: None, selected=w.V2_ARMS['controlled'], measured_cells=budget.B1_CELLS + ('32k-b4',), v2='controlled') assert fake.events == [] and record(result['decision'])['status'] == 'REJECT' result = w.window(FakeV2(), lambda row: None, selected=w.ARMS, measured_cells=m.CELLS, v2='production-policy') assert record(result['decision'])['status'] == 'REJECT' @pytest.mark.parametrize('env,fail,chain_fail,expect', [ ({'WINDOW_CHAIN': '1'}, '', False, 'chained'), ({'WINDOW_CHAIN': '1', 'WINDOW_CHAIN_LAST': '1'}, '', False, 'restored'), ({'WINDOW_CHAIN': '1'}, 'cell', False, 'restored'), ({'WINDOW_CHAIN': '1'}, '', True, 'restored-after-chain-failure')]) def test_v2_chain_end(monkeypatch: pytest.MonkeyPatch, env: dict[str, str], fail: str, chain_fail: bool, expect: str) -> None: for key in ('WINDOW_CHAIN', 'WINDOW_CHAIN_LAST'): monkeypatch.delenv(key, raising=False) for key, value in env.items(): monkeypatch.setenv(key, value) fake = FakeV2(fail, chain_fail=chain_fail) result = w.window(fake, lambda row: None, selected=w.V2_ARMS['controlled'], measured_cells=m.CELLS, v2='controlled') if expect == 'chained': assert fake.events[-1] == 'chain-close' and 'restore' not in fake.events assert result['chain'] == 'CHAINED' and record(result['restore'])['status'] == w.CHAINED assert record(result['decision'])['status'] == 'CANDIDATE' else: assert fake.events[-1] == 'restore' and result.get('chain') != 'CHAINED' assert ('chain-close' in fake.events) == (expect == 'restored-after-chain-failure') if expect == 'restored-after-chain-failure': assert 'quiet proof failed' in str(result['chain_close_error']) assert record(result['decision'])['status'] == 'CANDIDATE' # measurements stand; production restored if fail: assert record(result['decision'])['status'] == 'REJECT' def test_admission_and_pair_only_never_chain(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv('WINDOW_CHAIN', '1') fake = FakeV2() result = w.window(fake, lambda row: None, selected=()) assert fake.events[-1] == 'restore' and 'chain-close' not in fake.events and result['status'] == 'PAIR_QUALIFIED' def test_chained_start_failure_restores_with_production_stopped(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv('WINDOW_CHAIN', '1') fake = FakeV2('hold', down=True) result = w.window(fake, lambda row: None, selected=w.V2_ARMS['controlled'], measured_cells=m.CELLS, v2='controlled') assert fake.restored_with is True and result['chain_state_at_start'] == 'chained: production down' assert record(result['decision'])['status'] == 'REJECT' # ---- runner integration: real Live methods over fake transports def live(tmp_path: Path, plain: object | None = None, clock: base.Clock | None = None) -> w.Live: out = tmp_path / 'out' out.mkdir(exist_ok=True) sup = pf.Supervisor(out, clock=clock or base.Clock()) return w.Live(out, sup, native=prior.WORKER, remote='experiment', expected=prior.WORKER.commit, data_dir=tmp_path, tag='tag', admission={'verifier_sha256': {}}, plain=plain, clock=clock or base.Clock()) # type: ignore[arg-type] @pytest.mark.parametrize('fail', ['', 'jit', 'serving', 'stash', 'commit-close', 'placeholder-up', 'placeholder-proof']) def test_live_chain_close_order_and_marker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fail: str) -> None: monkeypatch.setenv('WINDOW_CHAIN', '1') monkeypatch.delenv('WINDOW_CHAIN_LAST', raising=False) monkeypatch.setattr(chain, 'MARKER', tmp_path / 'CHAINED') events: list[str] = [] def plain(label: str, argv: list[str], timeout: float) -> tuple[int, str]: events.append(label) if label.startswith('owned-fence-'): return 0, owner.FENCED if label.startswith(('owned-absent-', 'owned-reap-')): return 0, owner.ABSENT if label == 'pair-close': return 0, 'PRODUCTION HOLD OK' if label.startswith('jit-wait-'): return (1, '') if fail == 'jit' else (0, 'PP4 EXPERIMENT ABSENT') if label == 'chain-production-down': return 0, 'SERVING' if fail == 'serving' else 'DOWN' if label.startswith('chain-placeholder-up-'): assert argv[-1] == w.PLACEHOLDER_UP return (1, '') if fail == 'placeholder-up' else (0, '') if label.startswith('chain-placeholder-proof-'): return 0, 'PLACEHOLDER=0' if fail == 'placeholder-proof' else 'PLACEHOLDER=1' if label == 'chain-hold-stash': assert '-restore-close' in argv[-1] and 'rm -f' in argv[-1] return (1, '') if fail == 'stash' else (0, '') if label == 'commit-close': return (1, '') if fail == label else (0, owner.CLOSED) if label == 'cancel-close': return 0, 'SCOPED CLOSE COMMIT FENCED' if label == 'retain-close': return 0, owner.RETAINED if label == 'experiment-stop': return 0, '' raise AssertionError(label) runner = live(tmp_path, plain) runner.hold_written = True runner.owner_tags = ['tag-hold', 'tag-s1-A-worker', 'tag-s1-A-engine'] saved = signal.signal(signal.SIGTERM, signal.SIG_IGN) try: if fail: with pytest.raises(ValueError): runner.chain_close() assert not chain.MARKER.exists() assert ('retain-close' in events) == (fail == 'commit-close') assert 'start' not in events and 'check' not in events else: result = runner.chain_close() assert result['status'] == w.CHAINED and result['restore_deadline_met'] is True assert events.index('pair-close') < events.index('experiment-stop') < events.index('jit-wait-sparkhead') assert events.index('chain-production-down') < events.index('chain-hold-stash') < events.index('commit-close') # Default-socket placeholders go up before the first fence takes PP4's scoped placeholder down. assert events.index('chain-placeholder-up-sparkworker-wired') < min(i for i, e in enumerate(events) if e.startswith('owned-fence-')) assert events.index('chain-placeholder-proof-sparkworker-wired') < events.index('chain-hold-stash') assert events[-1] == 'cancel-close' and not any(e in events for e in ('start', 'check', 'close')) marker = chain.read_marker() assert marker is not None and record(marker['quiet_proof'])['production'] == 'DOWN' assert marker['placeholders'] == chain.PLACEHOLDERS_KEPT finally: signal.signal(signal.SIGTERM, saved) runner.sup.close() @pytest.mark.parametrize('state', ['production', 'chained', 'marker-without-chain', 'down-without-marker', 'not-quiet', 'no-placeholders', 'snapshot-changed']) def test_live_production_preflight_uses_the_chain_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, state: str) -> None: monkeypatch.setattr(chain, 'MARKER', tmp_path / 'CHAINED') previous = tmp_path / 'PREV-W' previous.mkdir() ranks = [base.prod_identity(0), base.prod_identity(1)] (previous / 'rollback.json').write_text(json.dumps({'production_commit': w.PROD_COMMIT, 'tag': w.PROD_TAG, 'ranks': ranks})) if state != 'down-without-marker' and state != 'production': chain.MARKER.write_text(json.dumps({'schema': chain.SCHEMA, 'window': str(previous), 'production_snapshot': chain.snapshot_ref(previous / 'rollback.json')})) if state == 'snapshot-changed': (previous / 'rollback.json').write_text(json.dumps({'production_commit': w.PROD_COMMIT, 'tag': w.PROD_TAG, 'ranks': ranks, 'edited': True})) if state == 'marker-without-chain': monkeypatch.delenv('WINDOW_CHAIN', raising=False) else: monkeypatch.setenv('WINDOW_CHAIN', '1') serving = state == 'production' calls: list[str] = [] def ssh(label: str, host: str, command: str, stdin: str | None = None, timeout: float = 120, inflight: int = 0) -> str: calls.append(label) if label == 'chain-state': assert command == w.CHAIN_PROBE return f'HOLD=0\nSERVING={int(serving)}\n' if label.startswith('chain-quiet-'): return '' if state == 'not-quiet' else 'PP4 CHAIN QUIET\n' if label.startswith('production-clock-'): assert '/etc/environment' in command return 'GPU_CLOCK_CAP=2300\n' if label.startswith('chain-placeholders-'): assert chain.PLACEHOLDER_PROBE in command return 'PLACEHOLDER=0\n' if state == 'no-placeholders' and host == 'sparkworker-wired' else 'PLACEHOLDER=1\n' return '' def as_json(label: str, host: str, command: str, stdin: str | None = None) -> dict[str, Json]: calls.append(label) return {'data': [{'max_model_len': 327680, 'root': '/srv/dsv41/legal-v2a'}]} runner = live(tmp_path) monkeypatch.setattr(runner, 'ssh', ssh) monkeypatch.setattr(runner, 'json', as_json) try: if state in ('marker-without-chain', 'down-without-marker', 'not-quiet', 'no-placeholders', 'snapshot-changed'): with pytest.raises(ValueError): runner.production_preflight() else: runner.production_preflight() assert runner.production_down() is (state == 'chained') assert runner.production_clock is not None and runner.production_clock['mhz'] == 2300 if state == 'production': assert 'prod-models' in calls and runner.carried is None assert not any(c.startswith('chain-placeholders-') for c in calls) # probed only with a marker else: assert 'prod-models' not in calls and calls[-2:] == ['chain-quiet-sparkhead', 'chain-quiet-sparkworker-wired'] assert ['chain-placeholders-sparkhead', 'chain-placeholders-sparkworker-wired'] == [c for c in calls if 'placeholders' in c] runner.snapshot() saved = json.loads((runner.out / 'rollback.json').read_text()) assert saved['carried_from'] == str(previous / 'rollback.json') and saved['ranks'] == ranks assert [r['id'] for r in runner.production] == ['prod0', 'prod1'] and runner.seen == {'prod0', 'prod1'} finally: runner.sup.close() def test_chained_production_identity_checks(tmp_path: Path) -> None: good = {'production_commit': w.PROD_COMMIT[:7], 'tag': w.PROD_TAG, 'ranks': [base.prod_identity(0), base.prod_identity(1)]} path = tmp_path / 'rollback.json' path.write_text(json.dumps(good)) assert w.chained_production(path)['tag'] == w.PROD_TAG for change in ({'tag': 'prod-k1-327-20261023'}, {'production_commit': '21bed31'}, {'ranks': [base.prod_identity(0)]}, {'ranks': [{'id': 'x'}, {'id': 'y'}]}): path.write_text(json.dumps({**good, **change})) with pytest.raises(ValueError): w.chained_production(path) @pytest.mark.skipif(not REAL_ROLLBACK.is_file(), reason='real PP4 window rollback.json absent') def test_real_pp4_window_rollback_is_a_valid_carried_identity() -> None: data = w.chained_production(REAL_ROLLBACK) assert data['production_commit'] == w.PROD_COMMIT and len(cast(list[Json], data['ranks'])) == 2 def test_runtime_production_policy_arm() -> None: prod = base.prod_identity() record(prod['env'])['VLLM_DSV41_ADAPTIVE_K'] = '1' spec = '{"method":"dspark","num_speculative_tokens":5,"num_speculative_tokens_per_batch_size":[[1,2,5],[3,4,4]]}' prod['argv'] = str(prod['argv']) + ' --speculative-config ' + shlex.quote(spec) overlay: dict[str, Json] = {pf.DIST + 'vllm/v1/worker/gpu/spec_decode/remote_dspark/client.py': 'e' * 64} arm = w.V2_ARMS['production-policy'][0] current = copy.deepcopy(prod) current['argv'] = str(current['argv']).replace('--port 8000', '--port 8001') current['env'] = {**record(prod['env']), 'VLLM_PP4_ROUND_DIR': '/pp4-round', 'VLLM_DSPARK_SAMPLED_TOP32': '0', **dict(zip(w.IDENTITY_KEYS, (prior.WORKER.identity, prior.WORKER.markov)))} current['mounted'] = {**record(prod['mounted']), **overlay} w.runtime(prod, current, arm, prior.WORKER, overlay) fixed = copy.deepcopy(current) fixed['argv'] = str(fixed['argv']).replace(shlex.quote(spec), shlex.quote(w.SPEC)) with pytest.raises(ValueError, match='production speculative config'): w.runtime(prod, fixed, arm, prior.WORKER, overlay) policy = copy.deepcopy(current) record(policy['env']).update(w.COMMON_ENV) with pytest.raises(ValueError, match='environment drift'): w.runtime(prod, policy, arm, prior.WORKER, overlay) block = '{"method":"dspark","num_speculative_tokens":5,"rejection_sample_method":"block"}' for target in (prod, current): target['argv'] = str(target['argv']).replace(shlex.quote(spec), shlex.quote(block)) with pytest.raises(ValueError, match='standard verification'): w.runtime(prod, current, arm, prior.WORKER, overlay) record(prod['env'])['VLLM_DSV41_ADAPTIVE_K'] = '0' with pytest.raises(ValueError, match='not adaptive'): w.runtime(prod, current, arm, prior.WORKER, overlay) def render(tmp_path: Path, config: str, role: str, production_env: bytes) -> tuple[int, str, str]: home = tmp_path / 'home' (home / 'dsv41-deploy').mkdir(parents=True, exist_ok=True) (home / 'dsv41-deploy' / 'dsv41.env').write_bytes(production_env) for name in ('vllm/v1/worker/gpu/model_runner.py', 'vllm/v1/worker/gpu/spec_decode/rejection_sampler_utils.py', 'vllm/v1/worker/gpu/dsv41_pp4_round_trace.py'): file = tmp_path / prereq.OVERLAY_REL / name file.parent.mkdir(parents=True, exist_ok=True) file.touch() (tmp_path / 'offload/launch/windows').mkdir(parents=True, exist_ok=True) for name in (w.CONFIG, w.PRODUCTION_CONFIG): (tmp_path / name).write_bytes((pf.ROOT / name).read_bytes()) command = ('set -eu; source ' + shlex.quote(config) + '; printf "%s\\n%s\\n%s\\n%s\\n%s\\n" "$EXP_DOCKER_ENV" ' '"$EXTRA_DOCKER_ARGS" "$OVERLAYS" "${SPEC_OVERRIDE-UNSET}" "$LABEL"') proc = subprocess.run(['bash', '-c', command], cwd=tmp_path, capture_output=True, text=True, check=False, env={**os.environ, 'HOME': str(home), 'PP4_ROLE': role, 'PP4_NATIVE_IDENTITY': prior.WORKER.identity, 'PP4_NATIVE_MARKOV_IDENTITY': prior.WORKER.markov, 'PP4_TRACE_RUN': 'test-trace'}) return proc.returncode, proc.stdout, proc.stderr def test_production_policy_config_differs_from_greedy_a_only_by_policy(tmp_path: Path) -> None: production_env = subprocess.check_output(['git', 'show', w.PROD_COMMIT + ':dsv41.env'], cwd=pf.ROOT) rc, a, err = render(tmp_path, w.CONFIG, 'A', production_env) assert rc == 0, err rc, p, err = render(tmp_path, w.PRODUCTION_CONFIG, 'P', production_env) assert rc == 0, err a_lines, p_lines = a.splitlines(), p.splitlines() policy = {'VLLM_DSV41_ADAPTIVE_K=0', 'VLLM_DSPARK_DEPTH_POLICY=fixed:5', 'VLLM_DSPARK_DEPTH_FILE=', 'VLLM_DSPARK_LOOKUP=', 'VLLM_DSPARK_REMOTE_WANT_MARGIN=0'} assert p_lines[0].split() == [t for t in a_lines[0].split() if t not in policy] assert policy <= set(a_lines[0].split()) and 'VLLM_DSPARK_SAMPLED_TOP32=0' in p_lines[0].split() assert p_lines[1:3] == a_lines[1:3] # same mounts and overlays assert a_lines[3] == w.SPEC and p_lines[3] == 'UNSET' and p_lines[4] == 'PP4-P-production-policy' rc, _, err = render(tmp_path, w.PRODUCTION_CONFIG, 'A', production_env) assert rc != 0 and 'role P required' in err seeded = production_env.replace(b'OFFLOAD_DSPARK_ENV="', b'OFFLOAD_DSPARK_ENV="VLLM_DSPARK_LOOKUP=x ', 1) rc, _, err = render(tmp_path, w.PRODUCTION_CONFIG, 'P', seeded) assert rc != 0 and 'VLLM_DSPARK_LOOKUP' in err def test_main_v2_refuses_before_any_remote_action(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: def forbidden(*args: object, **kwargs: object) -> None: raise AssertionError('remote/process action forbidden') monkeypatch.setattr(subprocess, 'Popen', forbidden) assert w.main(['--measure-v2', 'controlled']) == 0 assert 'PP4 v2 controlled' in capsys.readouterr().out monkeypatch.setenv('WINDOW_COORDINATOR', 'Deepseek/coordinator') common = ['--execute', '--qualification', str(tmp_path / 'q.json'), '--expected-experiment-commit', '2' * 40] assert w.main([*common, '--measure-v2', 'controlled']) == 1 assert '--rng-gate' in capsys.readouterr().err assert w.main([*common, '--rng-gate', str(tmp_path / 'g.json')]) == 1 assert 'belong to --measure-v2' in capsys.readouterr().err assert w.main([*common, '--measure-v2', 'controlled', '--admission', '--rng-gate', str(tmp_path / 'g.json')]) != 0 def test_v2_greedy_uses_the_round15_near_tie_path_and_cells_use_v2(tmp_path: Path) -> None: """P and B greedy checks run pp4_cells' greedy (R15: first differing token in the common prefix; only a pure prefix refuses); only the measured cells switch to the v2 driver.""" runner = live(tmp_path) try: runner.v2 = 'production-policy' assert runner.driver('greedy', tmp_path / 'g.json')[2] == 'offload.launch.windows.pp4_cells' assert runner.driver('cell', tmp_path / 'c.json')[2] == 'offload.launch.windows.pp4_measure_v2' assert runner.cell_timeout('32k-b1') == 35 * m.WAVES * 3 + 60 runner.v2 = None assert runner.driver('cell', tmp_path / 'c.json')[2] == 'offload.launch.windows.pp4_cells' finally: runner.sup.close() import inspect assert 'len(ta) == len(tb)' not in inspect.getsource(cells.near_tie) def test_production_cap_and_same_clock_guard() -> None: assert m.production_cap(['GPU_CLOCK_CAP=2300\n', 'GPU_CLOCK_CAP=2300\n']) == 2300 assert m.production_cap(['', '']) == m.DEFAULT_CAP_MHZ == 2150 assert m.production_cap(['GPU_CLOCK_CAP=\n', 'GPU_CLOCK_CAP=\n']) is None with pytest.raises(ValueError, match='disagree'): m.production_cap(['GPU_CLOCK_CAP=2300\n', '']) env = subprocess.check_output(['git', 'show', w.PROD_COMMIT + ':dsv41.env'], cwd=pf.ROOT, text=True) assert 'GPU_CLOCK_CAP=${GPU_CLOCK_CAP-2150}' in env names = ['s1-P', 's2-B', 's3-P'] at: dict[str, Json] = {n: {'spark': 2300, 'sparkworker': 2300} for n in names} assert m.clock_guard(at, names, [], {'mhz': 2300})['consistent'] is True off = m.clock_guard(at, names, [], {'mhz': 2150}) assert off['consistent'] is False and 'not production' in str(off['reason']) assert m.clock_guard({n: {} for n in names}, names, [], {'mhz': None})['consistent'] is True assert m.clock_guard({n: {} for n in names}, names, [], {'mhz': 2300})['consistent'] is False def test_archive_spread_uses_only_same_commit_same_clock_rows(tmp_path: Path) -> None: archive = tmp_path / 'archive.jsonl' rows = starts('controlled', candidate=WIN) for window, mhz, scale in (('A-W-20261027T1', 2300, 1.0), ('B-W-20261027T2', 2300, 1.0), ('C-W-20261027T3', 2150, 1.3)): clock: dict[str, Json] = {'per_start': {n: {'spark': mhz, 'sparkworker': mhz} for n in ('s1-A', 's2-B', 's3-A')}} for start in ('s1-A', 's3-A'): record(rows[start])['cells'] = {c: v2_cell(c, start, seconds=4. * scale) for c in m.CELLS} assert w.append_archive({'starts': rows}, 'controlled', window_name=window, experiment_commit='2' * 40, clock=clock, path=archive)['status'] == 'APPENDED' clock_now: dict[str, Json] = {**CLOCK_OK, 'per_start': {n: {'spark': 2300, 'sparkworker': 2300} for n in ('s1-A', 's2-B', 's3-A')}} result = m.decide(starts('controlled', candidate=WIN), 'controlled', clock=clock_now, rng=GATE, archive=archive, production_commit=w.PROD_COMMIT) for cell in m.CELLS: assert record(record(result['archive'])[cell])['archive_day_to_day_spread'] == 0.0 # the 2150 MHz window is excluded @pytest.mark.parametrize('case', ['chained-start-preflight-failure', 'chained-start-hold-never-written', 'no-marker']) def test_v2_chain_abort_when_production_was_left_down(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, case: str) -> None: """WIN2-R2 FIX 1: a chained window that stops without owning a HOLD runs the shared chain abort, never NOT NEEDED.""" monkeypatch.setattr(chain, 'MARKER', tmp_path / 'CHAINED') monkeypatch.setenv('WINDOW_CHAIN', '1') monkeypatch.delenv('WINDOW_CHAIN_LAST', raising=False) if case != 'no-marker': chain.MARKER.write_text(json.dumps({'schema': chain.SCHEMA, 'window': str(tmp_path)})) fake = FakeV2('preflight' if case != 'chained-start-hold-never-written' else 'hold', down=case == 'chained-start-hold-never-written') result = w.window(fake, lambda row: None, selected=w.V2_ARMS['controlled'], measured_cells=m.CELLS, v2='controlled') if case == 'no-marker': assert result['end_action'] == 'none' and 'chain-abort' not in fake.events assert record(result['restore'])['status'] == 'NOT NEEDED: preflight only' else: assert fake.events[-1] == 'chain-abort' and result['chain'] == 'ABORT' assert str(record(result['restore'])['status']).startswith('RESTORED: chain ended by abort') assert record(result['decision'])['status'] == 'REJECT' if case == 'chained-start-hold-never-written': assert 'own_restore' in result and fake.restored_with is True