| """CPU-only PP4 rehearsals. All serving, host, unit and GPU operations are fakes.""" |
| from __future__ import annotations |
|
|
| import ast |
| import copy |
| import importlib |
| import json |
| import os |
| import shlex |
| import signal |
| import subprocess |
| import sys |
| import time |
| from pathlib import Path |
| from typing import cast |
|
|
| import pytest |
|
|
| from offload.launch.windows import pf1_window as pf |
| from offload.launch.windows import pp4_cells as cells |
| from offload.launch.windows import pp4_metrics as metric |
| 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.prefill_maxlen.evidence import Json, record |
|
|
| WORKER = w.Worker('/opt/dsv41/x86host-data/pp4/source/repo', 'a' * 40, 'b' * 64, 'c' * 64) |
|
|
|
|
| def block_cell(name: str, sampled: bool = False, start: str = 'fixture') -> dict[str, Json]: |
| batch = cells.CELLS[name].batch |
| waves: list[Json] = [{'wave': wave, 'family': cells.FAMILIES[wave % 2], 'batch': batch, |
| 'spec_request_rounds': 1000.0, 'accepted_tokens': 1200.0 if sampled else 1000.0, |
| 'server_decode_seconds': 20.0, 'median_ttft_s': 1.0, 'completed_requests': batch, |
| 'streams': [{'prompt_sha256': name+str(wave)+'-'+str(i), |
| 'cache_salt_sha256': start+'-'+name+str(wave)+'-'+str(i), |
| 'sampling': {'temperature': 1, 'top_p': 1, 'seed': 42+wave, 'priority': 10, 'max_tokens': 512, 'ignore_eos': True}, |
| 'usage': {'completion_tokens': 512}, 'finish': 'length'} for i in range(batch)]} |
| for wave in range(metric.group_count(name))] |
| return {'cell': name, 'batch': batch, 'metric_registration': metric.REGISTRATION, 'waves': waves, |
| 'median_decode_tps': 110.0 if sampled else 100.0, |
| 'median_ttft_s': 1.0, 'p95_round_ms': 50.0, |
| 'complete_round_profile': {'by_family': {family: {'p95_round_ms': 50.0} for family in cells.FAMILIES}}, |
| 'streams': [s for row in waves for s in cast(list[Json], record(row)['streams'])]} |
|
|
|
|
|
|
| def complete() -> dict[str, Json]: |
| return {a.name: {'role': a.role, 'status': 'COMPLETE', 'greedy': {'identical': 6, 'near_tie': 0, 'checks': [{'i': i, 'identical': True} for i in range(6)]}, |
| 'cells': {name: block_cell(name, a.role == 'B', a.name) for name in cells.CELLS}} for a in w.ARMS} |
|
|
|
|
| def test_preregistered_rule_and_missing_ttft() -> None: |
| rows = complete() |
| result = w.decision(rows) |
| assert result['status'] == 'CANDIDATE FOR REVIEW' and result['source_families'] == 2 |
| assert result['resamples'] == 2000 and result['bootstrap_seed'] == 4103 |
| record(record(record(rows[w.ARMS[1].name])['cells'])['32k-b4'])['median_ttft_s'] = None |
| assert w.decision(rows)['status'] == 'INCONCLUSIVE' |
|
|
|
|
| @pytest.mark.parametrize('change', ['incomplete', 'different-input', 'ttft', 'server-cost', 'greedy', 'family-loss', 'low-gain']) |
| def test_rule_rejects_bad_evidence(change: str) -> None: |
| rows = complete() |
| if change == 'incomplete': |
| rows.pop(w.ARMS[-1].name) |
| for arm in (w.ARMS[1],): |
| block = record(rows[arm.name]) |
| cell = record(record(block['cells'])['short-b1']) |
| if change == 'different-input': |
| record(cast(list[Json], record(cast(list[Json], cell['waves'])[0])['streams'])[0])['prompt_sha256'] = 'different' |
| elif change == 'ttft': cell['median_ttft_s'] = 1.051 |
| elif change == 'server-cost': |
| for wave in cast(list[Json], cell['waves']): record(wave)['server_decode_seconds'] = 21.01 |
| elif change == 'greedy': record(cast(list[Json], record(block['greedy'])['checks'])[5])['identical'] = False |
| elif change == 'family-loss': |
| for name in cells.CELLS: |
| for wave in cast(list[Json], record(record(block['cells'])[name])['waves']): |
| if record(wave)['family'] == 'opinions': |
| record(wave)['accepted_tokens'] = 800.0 |
| elif change == 'low-gain': |
| for name in cells.CELLS: |
| for wave in cast(list[Json], record(record(block['cells'])[name])['waves']): |
| record(wave)['accepted_tokens'] = 1039.0 |
| if change in ('different-input', 'greedy'): |
| with pytest.raises(ValueError): w.decision(rows) |
| else: |
| assert w.decision(rows)['status'] != 'CANDIDATE FOR REVIEW' |
|
|
|
|
| def test_runtime_declares_exact_common_differences() -> None: |
| prod = base.prod_identity() |
| prod['argv'] = str(prod['argv']) + " --speculative-config '{\"method\":\"dspark\",\"num_speculative_tokens\":5,\"x\":1}'" |
| overlay: dict[str, Json] = {pf.DIST + 'vllm/v1/worker/gpu/spec_decode/remote_dspark/client.py': 'e' * 64} |
| for arm in w.ARMS: |
| live = copy.deepcopy(prod) |
| live['argv'] = str(live['argv']).replace('--port 8000', '--port 8001').replace( |
| '{"method":"dspark","num_speculative_tokens":5,"x":1}', w.SPEC) |
| live['env'] = {**record(prod['env']), **w.COMMON_ENV, |
| 'VLLM_DSPARK_SAMPLED_TOP32': str(int(arm.role == 'B')), |
| **dict(zip(w.IDENTITY_KEYS, (WORKER.identity, WORKER.markov)))} |
| live['mounted'] = {**record(prod['mounted']), **overlay} |
| w.runtime(prod, live, arm, WORKER, overlay) |
| record(live['env'])['NCCL_PROTO'] = 'drift' |
| with pytest.raises(ValueError, match='environment drift'): |
| w.runtime(prod, live, arm, WORKER, overlay) |
|
|
|
|
| class Fake: |
| def __init__(self, fail: str = '') -> None: |
| self.fail = fail |
| self.events: list[str] = [] |
|
|
| def preflight(self) -> None: |
| self.events.append('preflight') |
| if self.fail == 'preflight': raise ValueError('missing qualification') |
| def open_hold(self) -> float: |
| self.events.append('hold') |
| if self.fail == 'hold': raise ValueError('lost HOLD reply') |
| return 1.0 |
| def snapshot(self) -> None: self.events.append('snapshot') |
| def qualify_pair(self) -> dict[str, Json]: |
| self.events.append('pair-qualification') |
| if self.fail == 'pair-qualification': raise ValueError('physical peer refused') |
| return {'fixture': True} |
| def fits(self, arm: w.Arm) -> bool: return self.fail != 'time' |
| def start(self, arm: w.Arm) -> None: |
| self.events.append(arm.name) |
| if self.fail == 'start': raise ValueError('lost pair-open reply') |
| def build(self) -> None: self.events.append('freeze-inputs') |
| def warm(self, arm: w.Arm) -> None: self.events.append('warm-' + arm.name) |
| def greedy(self, arm: w.Arm) -> dict[str, Json]: |
| self.events.append('greedy-' + arm.name) |
| if self.fail == 'greedy': raise ValueError('greedy divergence') |
| return {'identical': 6, 'near_tie': 0, 'checks': [{'i': i, 'identical': True} for i in range(6)]} |
| def begin_measurement(self, arm: w.Arm) -> None: self.events.append('baseline-' + arm.name) |
| 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') |
| if self.fail == 'signal': raise KeyboardInterrupt('signal 15') |
| return block_cell(name, arm.role == 'B', arm.name) |
| def finish(self, arm: w.Arm) -> dict[str, Json]: return {'clean': True} |
| |
| def production_down(self) -> bool: return False |
| def measurement_context(self) -> dict[str, Json]: return {'clock': {'consistent': True}, 'rng_gate': None} |
| def chain_close(self) -> dict[str, Json]: raise AssertionError('chained close outside WINDOW_CHAIN=1') |
| def chain_abort(self) -> dict[str, Json]: raise AssertionError('chain abort without a chain marker') |
| def owns_hold(self) -> bool: return self.fail != 'hold' |
| def restore(self, *, production_stopped: bool) -> dict[str, Json]: |
| assert signal.getsignal(signal.SIGTERM) == signal.SIG_IGN |
| self.events.append('restore') |
| if self.fail == 'restore': raise ValueError('no absence proof') |
| return {'status': 'RESTORED: K1 START VERIFIED, greedy 6/6 IDENTICAL, PRODUCTION HOLD OK, window closed', |
| 'restore_deadline_met': True, 'production_stopped': production_stopped} |
|
|
|
|
| def test_full_off_on_off_rehearsal() -> None: |
| fake = Fake() |
| result = w.window(fake, lambda row: None) |
| assert result['status'] == 'COMPLETE' and record(result['decision'])['status'] == 'CANDIDATE FOR REVIEW' |
| assert [x for x in fake.events if x in [a.name for a in w.ARMS]] == ['s1-A', 's2-B', 's3-A'] |
| for arm in (w.ARMS[1],): |
| assert fake.events.index('greedy-' + arm.name) < next(i for i, x in enumerate(fake.events) if x.startswith('cell-' + arm.name)) |
| assert fake.events[-1] == 'restore' |
|
|
|
|
| @pytest.mark.parametrize('fail', ['hold', 'pair-qualification', 'start', 'time', 'greedy', 'cell', 'signal', 'restore']) |
| def test_every_failure_restores_and_cannot_promote(fail: str) -> None: |
| fake = Fake(fail) |
| result = w.window(fake, lambda row: None) |
| assert fake.events[-1] == 'restore' |
| assert record(result['decision'])['status'] == 'REJECT' |
| assert str(record(result['restore'])['status']).startswith('FAILED' if fail == 'restore' else 'RESTORED:') |
|
|
|
|
| def test_no_hold_on_failed_local_gate() -> None: |
| fake = Fake('preflight') |
| result = w.window(fake, lambda row: None) |
| assert fake.events == ['preflight'] and record(result['decision'])['status'] == 'REJECT' |
|
|
|
|
| @pytest.mark.parametrize('fail', ['', 'fence', 'reap', 'pair', 'start', 'greedy', 'deadline', 'jit', 'commit-close', 'cancel-close']) |
| def test_actual_live_restore_order(tmp_path: Path, fail: str) -> None: |
| out = tmp_path / 'out'; out.mkdir() |
| clock = base.Clock() |
| sup = pf.Supervisor(out, clock=clock) |
| events: list[str] = [] |
| live: w.Live |
| def plain(label: str, argv: list[str], timeout: float) -> tuple[int, str]: |
| assert signal.getsignal(signal.SIGTERM) == signal.SIG_IGN |
| events.append(label) |
| assert timeout <= w.RESTORE_S |
| if fail == 'deadline': clock.now += w.RESTORE_S + 1 |
| if label.startswith('owned-fence-'): |
| return (1, '') if fail == 'fence' else (0, owner.FENCED) |
| if label.startswith('owned-absent-'): |
| return 0, owner.ABSENT |
| if label.startswith('owned-reap-'): |
| return (1, '') if fail == 'reap' else (0, owner.ABSENT) |
| if label == 'pair-close': return (1, '') if fail == 'pair' else (0, 'PRODUCTION HOLD OK') |
| if label.startswith('jit-wait-'): return (1, '') if fail == 'jit' else (0, 'PP4 EXPERIMENT ABSENT') |
| if label == 'commit-close': return (1, '') if fail == label else (0, owner.CLOSED) |
| if label == 'cancel-close': return (1, '') if fail == label else (0, 'SCOPED CLOSE COMMIT FENCED') |
| if label == 'retain-close': return 0, owner.RETAINED |
| if label == 'experiment-stop': return 0, '' |
| text = base.RESTORE_OK[label] |
| if fail == 'start' and label == 'start': text = 'invented success' |
| if fail == 'greedy' and label == 'check': text = text.replace('5 IDENTICAL', '4 IDENTICAL') |
| return 0, text |
| live = w.Live(out, sup, native=WORKER, remote='experiment', expected=WORKER.commit, data_dir=tmp_path, |
| tag='tag', admission={'verifier_sha256': {}}, plain=plain, clock=clock) |
| live.hold_written = True |
| live.owner_tags = ['tag-hold', 'tag-s1-A-worker', 'tag-s1-A-engine'] |
| live.native_tags = ['tag-s1-A-worker'] |
| saved = signal.signal(signal.SIGTERM, signal.SIG_IGN) |
| try: |
| if fail: |
| with pytest.raises(ValueError): live.restore(production_stopped=True) |
| if fail not in ('commit-close', 'cancel-close'): |
| assert 'close' not in events |
| else: |
| assert 'retain-close' in events |
| else: |
| result = live.restore(production_stopped=True) |
| assert str(result['status']).startswith('RESTORED') |
| assert events.index('owned-reap-tag-s1-A-engine-sparkworker-wired') < events.index('pair-close') < events.index('start') |
| assert [e for e in events if e in ('start', 'check', 'close')] == ['start', 'check', 'close'] |
| assert events[-2:] == ['commit-close', 'cancel-close'] |
| for phase in ('tag-hold', 'tag-s1-A-worker', 'tag-s1-A-engine', 'tag-pair-close-1', |
| 'tag-restore-start-3', 'tag-restore-check-4', 'tag-restore-close'): |
| for host in w.OWNER_HOSTS: |
| assert f'owned-fence-{phase}-{host}' in events |
| assert f'owned-reap-{phase}-{host}' in events |
| finally: |
| signal.signal(signal.SIGTERM, saved) |
| sup.close() |
|
|
|
|
| def wait_file(path: Path) -> None: |
| until = time.monotonic() + 5 |
| while not path.exists(): |
| assert time.monotonic() < until, f'owned fake never reached {path.name}' |
| time.sleep(.01) |
|
|
|
|
| @pytest.mark.parametrize('pause', ['before-registration', 'pair-active']) |
| def test_exact_worker_late_pair_open_fenced(tmp_path: Path, pause: str) -> None: |
| home = tmp_path / 'home'; home.mkdir() |
| bin_dir = home / 'bin'; bin_dir.mkdir() |
| (bin_dir / 'python3').symlink_to(sys.executable) |
| (bin_dir / 'git').write_text('#!/bin/bash\nif [ "$1" = rev-parse ]; then echo ' + WORKER.commit + '; fi\n') |
| (bin_dir / 'git').chmod(0o700) |
| run = home / 'dsv41-offload'; run.mkdir() |
| (run / 'prod_window.sh').write_text('#!/bin/bash\nset -eu\necho inactive > "$HOME/state"\n' |
| 'touch "$HOME/pair-active"\nwhile [ ! -e "$HOME/release" ]; do sleep .01; done\n' |
| 'touch "$HOME/PAIR_OK"\n') |
| local_worker = w.Worker(str(home / 'source/repo'), WORKER.commit, WORKER.identity, WORKER.markov) |
| worker_tree = Path(local_worker.tree) / 'offload/dspark/worker/run'; worker_tree.mkdir(parents=True) |
| (worker_tree / 'gpu_locked.sh').write_text('#!/bin/sh\nexec sleep 30\n') |
| (home / 'state').write_text('active') |
| tag = 'pp4test-owned-cpu-pair' |
| script = w.worker_script(local_worker, tag, True) |
| launch = owner.command('launch', tag, ['bash', '-c', script], deadline=time.time()+30) |
| if pause == 'before-registration': |
| launch = 'touch "$HOME/pending"; while [ ! -e "$HOME/release" ]; do sleep .01; done\n' + launch |
| env = {**os.environ, 'HOME': str(home), 'PATH': str(bin_dir) + ':' + os.environ['PATH'], 'CUDA_VISIBLE_DEVICES': ''} |
| proc = subprocess.Popen(['bash', '-c', launch], env=env, start_new_session=True, |
| stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
| try: |
| wait_file(home / ('pending' if pause == 'before-registration' else 'pair-active')) |
| for mode in ('fence', 'reap', 'absent'): |
| result = subprocess.run(['bash', '-c', owner.command(mode, tag)], env=env, capture_output=True, text=True, timeout=30, check=False) |
| assert result.returncode == 0, result.stderr |
| (home / 'state').write_text('active') |
| (home / 'PAIR_OK').unlink(missing_ok=True) |
| (home / 'release').touch() |
| proc.wait(timeout=5) |
| assert (home / 'state').read_text() == 'active' and not (home / 'PAIR_OK').exists() |
| retry = subprocess.run(['bash', '-c', owner.command('launch', tag, ['bash', '-c', script], deadline=time.time()+30)], env=env, |
| capture_output=True, text=True, timeout=5, check=False) |
| assert retry.returncode == 75 |
| finally: |
| if proc.poll() is None: os.killpg(proc.pid, signal.SIGKILL) |
| proc.wait(timeout=5) |
| for mode in ('fence', 'reap', 'absent'): |
| subprocess.run(['bash', '-c', owner.command(mode, tag)], env=env, capture_output=True, timeout=30, check=True) |
|
|
|
|
| def test_dry_and_missing_admission_make_no_calls(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: |
| def forbidden(*args: object, **kwargs: object) -> None: raise AssertionError('remote/process action forbidden') |
| monkeypatch.setattr(subprocess, 'Popen', forbidden) |
| assert w.main([]) == 0 |
| assert capsys.readouterr().out.startswith('DRY RUN') |
| monkeypatch.setenv('WINDOW_COORDINATOR', 'Deepseek/coordinator') |
| assert w.main(['--execute']) == 1 |
| assert 'qualification' in capsys.readouterr().err |
|
|
|
|
| def test_admission_requires_every_primary_evidence_hash(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| evidence = tmp_path / 'evidence.json'; evidence.write_text('{}') |
| data: dict[str, Json] = {'schema': 'pp4-live-admission-v1', 'verdict': 'QUALIFIED', 'source_commit': WORKER.commit, |
| 'depth': 5, 'worker_identity': WORKER.identity, 'markov_identity': WORKER.markov, |
| 'runtime_python': prereq.PYTHON, 'runtime_torch': '2.12.1+cu130', |
| 'runtime_artifact_sha256': {prereq.STOCK: 'd' * 64, prereq.FACC: 'e' * 64, |
| 'p2p/' + pf.digest(pf.ROOT / 'offload/dspark/worker/p2p/p2p.cu')[:16] + '/dspark_p2p.so': 'f' * 64}, |
| 'gates': {g: {'pass': True, 'evidence_path': str(evidence), 'sha256': pf.digest(evidence)} for g in prereq.GATES}, |
| 'verifier_sha256': {'vllm/v1/worker/gpu/model_runner.py': 'a' * 64, |
| 'vllm/v1/worker/gpu/spec_decode/rejection_sampler_utils.py': 'b' * 64, |
| 'vllm/v1/worker/gpu/dsv41_pp4_round_trace.py': 'c' * 64}} |
| from offload.launch.windows import pp4_budget |
| monkeypatch.setattr(pp4_budget, 'admitted', lambda path, expected: {'block_budget_s': {'s1-A': 100, 's2-B': 100, 's3-A': 100}}) |
| path = tmp_path / 'receipt.json'; path.write_text(json.dumps(data)) |
| assert prereq.qualification(path, WORKER.commit)['verdict'] == 'QUALIFIED' |
| evidence.write_text('{"changed":true}') |
| with pytest.raises(ValueError, match='changed'): prereq.qualification(path, WORKER.commit) |
|
|
|
|
| def test_generated_scripts_compile_and_worker_is_production_k2() -> None: |
| for sampled in (False, True): |
| script = w.worker_script(WORKER, 'test-tag', sampled) |
| subprocess.run(['bash', '-n', '-c', script], check=True) |
| assert 'prod_window.sh pair COORD' in script and f'DSPARK_SAMPLED_TOP32={int(sampled)}' in script |
| assert '--tp 2 --root /opt/dsv41/x86host-data/dspark --max-model-len 1048576' in script |
| assert 'native-hybrid' not in script and 'DSPARK_WORKER_TP_EMULATE' in script |
| ast.parse(w.native_probe_script(WORKER, 'test-tag')) |
| subprocess.run(['bash', '-n', str(pf.ROOT / w.CONFIG)], check=True) |
|
|
|
|
| def test_decode_censor_is_explicit(monkeypatch: pytest.MonkeyPatch) -> None: |
| values = iter([0., 121.]) |
| monkeypatch.setattr(time, 'monotonic', lambda: next(values)) |
| import threading |
| first = threading.Event(); first.set() |
| with pytest.raises(TimeoutError, match='CENSORED'): |
| list(cells.decode_bound([b'a', b'b'], first)) |
|
|
|
|
| def test_actual_config_arms_differ_only_sample_switch(tmp_path: Path) -> None: |
| home = tmp_path / 'home'; prod = home / 'dsv41-deploy'; prod.mkdir(parents=True) |
| |
| |
| production_env = subprocess.check_output(['git', 'show', w.PROD_COMMIT + ':dsv41.env'], cwd=pf.ROOT) |
| (prod / '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() |
| config = pf.ROOT / w.CONFIG |
| results = [] |
| for role in ('A', 'B'): |
| command = 'set -eu; source ' + shlex.quote(str(config)) + '; printf "%s\\n%s\\n%s\\n%s\\n" "$EXP_DOCKER_ENV" "$EXTRA_DOCKER_ARGS" "$OVERLAYS" "$SPEC_OVERRIDE"' |
| proc = subprocess.run(['bash', '-c', command], cwd=tmp_path, |
| env={**os.environ, 'HOME': str(home), 'PP4_ROLE': role, 'PP4_NATIVE_IDENTITY': WORKER.identity, |
| 'PP4_NATIVE_MARKOV_IDENTITY': WORKER.markov, 'PP4_TRACE_RUN': 'test-trace'}, capture_output=True, text=True, check=False) |
| assert proc.returncode == 0, proc.stderr |
| results.append(proc.stdout) |
| assert results[0].replace('VLLM_DSPARK_SAMPLED_TOP32=0', 'VLLM_DSPARK_SAMPLED_TOP32=1') == results[1] |
| assert 'VLLM_ENGRAM_REMOTE_REQUIRED=0' in results[0] and 'VLLM_DSV41_ADAPTIVE_K=0' in results[0] |
| nested = pf.DIST + 'vllm/v1/worker/gpu/spec_decode/remote_dspark' |
| assert '/usr/local/lib/python3.12/dist-packages/remote_dspark' not in results[0] |
| env_lines = results[0].splitlines() |
| replacements = dict(pair.split('=', 1) for pair in env_lines[2].split()) |
| assert replacements['offload/dspark/spark/remote_dspark'] == nested |
| mounts = shlex.split(env_lines[1]) |
| destinations = [mounts[i+1].split(':')[1] for i in range(0, len(mounts), 2)] |
| assert nested not in destinations |
| assert len(destinations) == len(set(destinations)) |
| |
| original = subprocess.check_output(['bash', '-c', '. "$HOME/dsv41-deploy/dsv41.env"; printf %s "$OFFLOAD_DOCKER_ARGS"'], |
| env={**os.environ, 'HOME': str(home)}, text=True) |
| originals = shlex.split(original) |
| removed = {nested, pf.DIST+'vllm/v1/worker/gpu/model_runner.py', |
| pf.DIST+'vllm/v1/worker/gpu/spec_decode/rejection_sampler_utils.py'} |
| for i in range(0, len(originals), 2): |
| mount = originals[i+1] |
| if mount.split(':')[1] not in removed: |
| assert mount in mounts |
| assert pf.DIST+'vllm/v1/worker/gpu/spec_decode/__init__.py' in destinations |
| assert len({*destinations, *replacements.values()}) == len(destinations)+len(replacements) |
| pins = prereq.overlay_pins({'verifier_sha256': {}}) |
| assert pins and all(path.startswith(nested + '/') for path in pins) |
| assert nested+'/speculator.py' in pins |
|
|
|
|
|
|
| def test_complete_round_summary_matches_both_ranks_and_actual_batch() -> None: |
| ranks: dict[int, list[dict[str, Json]]] = {rank: [ |
| {'schema': 'pp4-complete-round-v1', 'cell': '32k-b4', 'rank': rank, 'round': i, |
| 'batch': 4, 'request_hash': 'same', 'failed': False, 'complete_wall_ms': float(i + rank + 1), |
| 'cuda_elapsed_ms': float(i + 1)} for i in range(20)] for rank in (0, 1)} |
| result = w.round_summary(ranks, '32k-b4', 4) |
| assert result['p95_round_ms'] == 20.0 and result['profiled_rounds'] == 20 |
| ranks[1].pop() |
| with pytest.raises(ValueError, match='both ranks'): w.round_summary(ranks, '32k-b4', 4) |
| ranks[0].pop(3); ranks[1].pop(3) |
| with pytest.raises(ValueError, match='missing complete-round'): w.round_summary(ranks, '32k-b4', 4) |
|
|
|
|
| def test_profile_helper_completes_real_round_boundary_with_fake_events(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| from types import SimpleNamespace |
|
|
| from offload.launch.windows import pp4_round_trace as trace |
| class Event: |
| def record(self) -> None: pass |
| def synchronize(self) -> None: pass |
| def elapsed_time(self, other: trace.Event) -> float: return 5.0 |
| monkeypatch.setenv('VLLM_PP4_ROUND_DIR', str(tmp_path)) |
| monkeypatch.setattr(importlib, 'import_module', lambda name: SimpleNamespace(cuda=SimpleNamespace(Event=lambda **kw: Event()), distributed=SimpleNamespace(get_rank=lambda: 0))) |
| runner = SimpleNamespace(parallel_config=SimpleNamespace(rank=0), execute_model_state=object()) |
| scheduler = SimpleNamespace(num_scheduled_tokens={'id': 6}, scheduled_spec_decode_tokens={'id': [1] * 5}) |
| trace.begin(runner, scheduler, False) |
| assert not (tmp_path / 'rank0.jsonl').exists(), 'disabled headline must not create CUDA events or traces' |
| (tmp_path / 'ACTIVE').write_text('32k-b1') |
| trace.begin(runner, scheduler, False) |
| trace.finish(runner) |
| row = json.loads((tmp_path / 'rank0.jsonl').read_text()) |
| assert row['cuda_elapsed_ms'] == 5 and row['batch'] == 1 and row['failed'] is False |
| assert 'id' not in row and row['request_hash'] != 'id' |
|
|
|
|
| def test_round_overlay_preserves_probability_abi_and_wraps_both_methods() -> None: |
| from offload.launch.windows.pp4_prepare import instrument |
| text = 'REMOTE_DRAFT_PROBS_ABI = "pp3-dyadic24-standard-v1"\nclass Runner:\n def execute_model(self): pass\n def sample_tokens(self): pass\n' |
| result = instrument(text) |
| tree = ast.parse(result) |
| assert 'REMOTE_DRAFT_PROBS_ABI = "pp3-dyadic24-standard-v1"' in result |
| cls = next(x for x in tree.body if isinstance(x, ast.ClassDef)) |
| assert all(isinstance(f, ast.FunctionDef) and len(f.decorator_list) == 1 for f in cls.body) |
| with pytest.raises(ValueError, match='anchors changed'): instrument('class Runner: pass\n') |
|
|
|
|
| @pytest.mark.parametrize('fail', [False, True]) |
| def test_profile_collection_uses_complete_rounds_and_clears_markers(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fail: bool) -> None: |
| out = tmp_path / 'out'; out.mkdir() |
| sup = pf.Supervisor(out) |
| live = w.Live(out, sup, native=WORKER, remote='experiment', expected=WORKER.commit, data_dir=tmp_path, |
| tag='owned', admission={'verifier_sha256': {}}) |
| live.submit = time.time() + 3600; live.block_deadline = live.submit; sup.abort_at = live.submit |
| events: list[str] = [] |
| monkeypatch.setattr(live, 'stats', lambda label: counter_stats()) |
| live.begin_measurement(w.ARMS[0]) |
| def call(label: str, argv: list[str], *, timeout: float = 120, inflight: int = 0) -> str: |
| events.append(label) |
| if '--output' not in argv: |
| return owner.FENCED if label.startswith('owned-fence-') else owner.ABSENT |
| if fail: raise TimeoutError('profile censored') |
| Path(argv[argv.index('--output') + 1]).write_text('{}') |
| return '' |
| def ssh(label: str, host: str, command: str, stdin: str | None = None, *, timeout: float = 120) -> str: |
| events.append(label) |
| if '-round-trace-' in label: |
| rank = pf.HOSTS.index(host) |
| return ''.join(json.dumps({'schema': 'pp4-complete-round-v1', 'cell': 'opinions:short-b1', |
| 'rank': rank, 'round': i, 'batch': 1, 'request_hash': 'same', 'failed': False, |
| 'complete_wall_ms': 20.0, 'cuda_elapsed_ms': 18.0}) + '\n' for i in range(20)) |
| return '' |
| def plain(label: str, argv: list[str], timeout: float) -> tuple[int, str]: |
| events.append(label) |
| return 0, owner.FENCED if label.startswith('owned-fence-') else owner.ABSENT |
| monkeypatch.setattr(live, 'call', call); monkeypatch.setattr(live, 'ssh', ssh); monkeypatch.setattr(live, 'plain', plain) |
| try: |
| if fail: |
| with pytest.raises(TimeoutError): live.profile_family(w.ARMS[0], 'short-b1', 'opinions', 0) |
| else: |
| result = live.profile_family(w.ARMS[0], 'short-b1', 'opinions', 0) |
| assert result['p95_round_ms'] == 20.0 |
| assert [x for x in events if x.startswith(arm_prefix := w.ARMS[0].name + '-profile-off-opinions-')] == [arm_prefix+h for h in pf.HOSTS] |
| finally: |
| sup.close() |
|
|
|
|
| @pytest.mark.parametrize('rank', [0, 1]) |
| def test_nested_package_mount_hash_drift_rejected_on_each_rank(rank: int) -> None: |
| prod = base.prod_identity() |
| prod['argv'] = str(prod['argv']) + " --speculative-config '{}'" |
| overlay = prereq.overlay_pins({'verifier_sha256': {}}) |
| live = copy.deepcopy(prod) |
| live['argv'] = str(live['argv']).replace('--port 8000', '--port 8001').replace("'{}'", shlex.quote(w.SPEC)) |
| live['env'] = {**record(prod['env']), **w.COMMON_ENV, 'VLLM_DSPARK_SAMPLED_TOP32': '0', |
| **dict(zip(w.IDENTITY_KEYS, (WORKER.identity, WORKER.markov)))} |
| live['mounted'] = {**record(prod['mounted']), **overlay} |
| w.runtime(prod, live, w.ARMS[0], WORKER, overlay) |
| path = pf.DIST+'vllm/v1/worker/gpu/spec_decode/remote_dspark/speculator.py' |
| record(live['mounted'])[path] = 'rank'+str(rank)+' drift' |
| with pytest.raises(ValueError): |
| w.runtime(prod, live, w.ARMS[0], WORKER, overlay) |
|
|
|
|
| def counter_stats(failed: int = 1, some: int = 2, ok: int = 100) -> dict[str, Json]: |
| return {'counters': {'rounds_ok': ok, 'rounds_failed': failed, 'rounds_some_failed': some, |
| 'fallback_internal_error': 1}, 'proxy_torch_loaded': False, |
| 'policy': 'capacity', 'async': True, 'async_proxy': {'mode': 'process'}, |
| 'async_handoff': {'mlock': 'ok'}, 'async_watchdog': {'fires': 0, 'running': True}} |
|
|
|
|
| def test_baseline_after_warmup_and_greedy_before_headline() -> None: |
| fake = Fake(); result = w.window(fake, lambda row: None) |
| assert result['status'] == 'COMPLETE' |
| for arm in w.ARMS: |
| baseline = fake.events.index('baseline-'+arm.name) |
| assert fake.events.index('warm-'+arm.name) < baseline |
| if arm.role == 'B': |
| assert fake.events.index('greedy-'+arm.name) < baseline |
| assert baseline < next(i for i, event in enumerate(fake.events) if event.startswith('cell-'+arm.name)) |
|
|
|
|
| @pytest.mark.parametrize('change', ['unchanged', 'headline', 'profile', 'reset', 'missing', 'late', 'stop-growth']) |
| def test_measurement_baseline_failure_growth_and_final_snapshot(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, |
| change: str) -> None: |
| sup = pf.Supervisor(tmp_path) |
| live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit, |
| data_dir=tmp_path, tag='counter', admission={'verifier_sha256': {}}) |
| current = counter_stats() |
| monkeypatch.setattr(live, 'stats', lambda label: copy.deepcopy(current)) |
| arm = w.ARMS[0] |
| try: |
| with pytest.raises(ValueError, match='baseline absent'): |
| live.measured_stats(arm, 'before-baseline') |
| live.begin_measurement(arm) |
| with pytest.raises(ValueError, match='cannot be renewed'): |
| live.begin_measurement(arm) |
| cs = record(current['counters']); cs['rounds_ok'] = 110 |
| live.measured_stats(arm, 'headline-before') |
| if change in ('headline', 'profile', 'late'): cs['rounds_failed'] = 2 |
| if change == 'reset': cs['rounds_ok'] = 1 |
| if change == 'missing': cs.pop('rounds_some_failed') |
| if change == 'stop-growth': cs['fallback_internal_error'] = 2 |
| if change == 'unchanged': |
| assert live.measured_stats(arm, 'headline-after')['counters'] == current['counters'] |
| assert live.measured_stats(arm, 'profile-after')['counters'] == current['counters'] |
| |
| |
| live.ids = ['id0', 'id1']; sup.arm_minima = {h: 1 for h in pf.HOSTS} |
| monkeypatch.setattr(live, 'ssh', lambda label, host, command: live.ids[pf.HOSTS.index(host)]) |
| monkeypatch.setattr(live, 'json', lambda *args: {}) |
| monkeypatch.setattr(pf, 'logs', lambda *args: {}) |
| assert record(live.finish(arm)['proxy'])['counters'] == current['counters'] |
| elif change == 'late': |
| with pytest.raises(ValueError, match='grew after measurement'): |
| live.finish(arm) |
| else: |
| with pytest.raises(ValueError): |
| live.measured_stats(arm, change+'-after') |
| finally: |
| sup.close() |
|
|
