| """CPU fakes only: no HTTP server, SSH, production endpoint or GPU is used.""" |
| from __future__ import annotations |
|
|
| from collections.abc import Iterator |
| import hashlib |
| from pathlib import Path |
| import subprocess |
| import urllib.request |
| import shlex |
| import signal |
|
|
| import pytest |
|
|
| from offload.launch.windows import post_close_watch as w |
| from offload.launch.windows import scoped_remote as owner |
|
|
| CHECK = ''.join(f'greedy K1-post vs restored: {i} IDENTICAL 100\n' for i in range(6)) + 'PRODUCTION HOLD OK\nCHECK: DONE\n' |
| W8 = {'start': 'START: K1 START VERIFIED\n', 'check': CHECK, 'close': 'CLOSE: window closed\n'} |
|
|
|
|
| class Clock: |
| def __init__(self) -> None: |
| self.value = 0.0 |
|
|
| def __call__(self) -> float: |
| return self.value |
|
|
| def sleep(self, amount: float) -> None: |
| self.value += amount |
|
|
|
|
| class Health: |
| def __init__(self, failures: set[int]) -> None: |
| self.failures = failures |
| self.calls = 0 |
|
|
| def probe(self) -> None: |
| self.calls += 1 |
| if self.calls in self.failures: |
| raise OSError('fake production unavailable') |
|
|
|
|
| class Recovery: |
| def __init__(self, fail: bool = False) -> None: |
| self.calls: list[float] = [] |
| self.fail = fail |
|
|
| def recover(self, deadline: float) -> None: |
| self.calls.append(deadline) |
| if self.fail: |
| raise w.WatchError('fake missing greedy index') |
|
|
|
|
| def test_schedule_is_fifteen_minutes_every_thirty_seconds() -> None: |
| clock, health, recovery = Clock(), Health(set()), Recovery() |
| logs: list[str] = [] |
| result = w.watch(health, recovery, log=logs.append, clock=clock, sleep=clock.sleep) |
| assert result == w.Result('PASS', 30, 0, 0) |
| assert clock() == 900 |
| assert not recovery.calls |
| assert logs[0].startswith('WATCH START:') and logs[-1].startswith('WATCH COMPLETE:') |
|
|
|
|
| def test_success_resets_failure_streak() -> None: |
| clock, health, recovery = Clock(), Health({1, 3, 5}), Recovery() |
| result = w.watch(health, recovery, log=lambda _: None, clock=clock, sleep=clock.sleep) |
| assert result.recoveries == 0 and result.failures == 3 |
|
|
|
|
| def test_two_consecutive_failures_recover_once_and_recurrence_is_loud() -> None: |
| clock, health, recovery = Clock(), Health({2, 3, 6, 7}), Recovery() |
| logs: list[str] = [] |
| result = w.watch(health, recovery, log=logs.append, clock=clock, sleep=clock.sleep) |
| assert result.status.startswith('FAILED: recurrence') |
| assert result.recoveries == 1 and len(recovery.calls) == 1 |
| assert recovery.calls == [60 + w.RECOVERY_S] |
| assert any('allowance exhausted' in line for line in logs) |
|
|
|
|
| def test_failed_recovery_is_never_retried() -> None: |
| clock, health, recovery = Clock(), Health({1, 2, 3}), Recovery(True) |
| logs: list[str] = [] |
| result = w.watch(health, recovery, log=logs.append, clock=clock, sleep=clock.sleep) |
| assert result == w.Result('FAILED: recovery unverified; retain HOLD', 2, 2, 1) |
| assert len(recovery.calls) == 1 |
| assert any('HOLD stays open' in line for line in logs) |
|
|
|
|
| @pytest.mark.parametrize('text', [ |
| CHECK.replace('5 IDENTICAL', '4 IDENTICAL'), |
| CHECK + 'greedy K1-post vs restored: 0 IDENTICAL 100\n', |
| CHECK.replace('5 IDENTICAL', '6 IDENTICAL'), |
| CHECK.replace('5 IDENTICAL 100', '5 DIVERGES at char 1'), |
| CHECK.replace('5 IDENTICAL 100', '5 IDENTICAL 100 junk'), |
| CHECK.replace('PRODUCTION HOLD OK', ''), |
| CHECK.replace('CHECK: DONE', ''), |
| CHECK + 'STOP: wrong KV size\n', |
| ]) |
| def test_strict_greedy_six_distinct_indices(text: str) -> None: |
| w.restore_verified('check', 0, CHECK) |
| with pytest.raises(w.WatchError): |
| w.restore_verified('check', 0, text) |
|
|
|
|
| @pytest.mark.parametrize('step', ['start', 'check', 'close']) |
| def test_w8_status_and_success_markers_required(step: str) -> None: |
| w.restore_verified(step, 0, W8[step]) |
| with pytest.raises(w.WatchError): |
| w.restore_verified(step, 1, W8[step]) |
| with pytest.raises(w.WatchError): |
| w.restore_verified(step, 0, '') |
|
|
|
|
| def test_probe_checks_models_and_sends_one_token_greedy_priority_twenty() -> None: |
| calls: list[tuple[str, dict[str, object] | None]] = [] |
| def request(path: str, body: dict[str, object] | None) -> object: |
| calls.append((path, body)) |
| if body is None: |
| return {'data': [{'root': w.MODEL_ROOT, 'id': 'legal-v2a', 'max_model_len': 327680}]} |
| return {'choices': [{'text': 'OK', 'finish_reason': 'length'}], 'usage': {'completion_tokens': 1}} |
| w.ProductionHealth(request=request).probe() |
| assert [path for path, _ in calls] == ['/v1/models', '/v1/completions'] |
| assert calls[1][1] == {'model': 'legal-v2a', 'prompt': 'Reply OK.', 'max_tokens': 1, |
| 'temperature': 0, 'top_p': 1, 'seed': 42, 'priority': 20, 'stream': False} |
|
|
|
|
| @pytest.mark.parametrize('reply', [ |
| {'data': []}, {'data': [{'root': '/wrong', 'id': 'x'}]}, {'data': 'wrong'}, |
| {'data': [{'root': w.MODEL_ROOT, 'id': 'model', 'max_model_len': 524288}]}, |
| {'data': [{'root': w.MODEL_ROOT, 'id': 'model'}]}, |
| ]) |
| def test_wrong_model_never_gets_completion_probe(reply: object) -> None: |
| calls: list[str] = [] |
| def request(path: str, body: dict[str, object] | None) -> object: |
| calls.append(path) |
| return reply |
| with pytest.raises(w.WatchError): |
| w.ProductionHealth(request=request).raw() |
| assert calls == ['/v1/models'] |
|
|
|
|
| @pytest.mark.parametrize('reply', [ |
| {'choices': [], 'usage': {'completion_tokens': 1}}, |
| {'choices': [{'text': 'OK', 'finish_reason': None}], 'usage': {'completion_tokens': 1}}, |
| {'choices': [{'text': 'OK', 'finish_reason': 'length'}], 'usage': {'completion_tokens': 0}}, |
| {'choices': [{'text': 'OK', 'finish_reason': 'length'}], 'usage': {'completion_tokens': True}}, |
| ]) |
| def test_incomplete_completion_is_health_failure(reply: object) -> None: |
| def request(path: str, body: dict[str, object] | None) -> object: |
| return {'data': [{'root': w.MODEL_ROOT, 'id': 'model', 'max_model_len': 327680}]} if body is None else reply |
| with pytest.raises(w.WatchError): |
| w.ProductionHealth(request=request).raw() |
|
|
|
|
| class Transport: |
| """Commands are parsed as argv via shlex; never executed.""" |
| def __init__(self, failure: str = '') -> None: |
| self.calls: list[tuple[str, list[str], float]] = [] |
| self.failure = failure |
|
|
| def __call__(self, label: str, argv: list[str], timeout: float) -> tuple[int, str]: |
| self.calls.append((label, argv, timeout)) |
| if label == self.failure: |
| return 1, 'fake failed transport' |
| if label == 'run-hold-open': |
| return 0, 'window open: POST-CLOSE-WATCH fake:\n' |
| if label == 'run-stop': |
| return 0, '' |
| if label.startswith('quiet-'): |
| return 0, 'WATCH SPARK QUIET\n' |
| if label.startswith('run-restore-'): |
| return 0, W8[label.removeprefix('run-restore-')] |
| for prefix, marker in (('fence-', owner.FENCED), ('reap-', owner.ABSENT), ('absent-', owner.ABSENT)): |
| if label.startswith(prefix): |
| return 0, marker + '\n' |
| return 0, {'commit-close': owner.CLOSED, 'cancel-close': 'SCOPED CLOSE COMMIT FENCED', |
| 'retain-close': owner.RETAINED}[label] + '\n' |
|
|
|
|
| @pytest.fixture |
| def recovery(monkeypatch: pytest.MonkeyPatch) -> Iterator[tuple[w.ScopedRecovery, Transport]]: |
| transport = Transport() |
| |
| monkeypatch.setattr(Path, 'read_bytes', lambda _: b'reviewed W8') |
| monkeypatch.setattr(w, 'RESTORE_SHA256', hashlib.sha256(b'reviewed W8').hexdigest()) |
| backend = w.ScopedRecovery('fake', Path('/unused/not-created'), call=transport, clock=lambda: 0, |
| wall=lambda: 1000) |
| yield backend, transport |
|
|
|
|
| def test_recovery_owns_all_phases_and_proves_all_hosts_before_close( |
| recovery: tuple[w.ScopedRecovery, Transport]) -> None: |
| backend, transport = recovery |
| backend.recover(1800) |
| labels = [label for label, _, _ in transport.calls] |
| assert [label for label in labels if label.startswith('run-')] == [ |
| 'run-hold-open', 'run-stop', 'run-restore-start', 'run-restore-check', 'run-restore-close'] |
| assert labels.index('quiet-sparkhead') < labels.index('run-restore-start') |
| quiet = next(argv[-1] for label, argv, _ in transport.calls if label == 'quiet-sparkhead') |
| assert 'docker' in quiet and 'inspect' in quiet and '{{.ID}}' in quiet |
| assert 'tf_experts' in quiet and '_load_engram_remote_ext' in quiet |
| assert '--query-compute-apps=pid' in quiet |
| opened = next(argv[-1] for label, argv, _ in transport.calls if label == 'run-hold-open') |
| stop = next(argv[-1] for label, argv, _ in transport.calls if label == 'run-stop') |
| assert 'os.link(candidate, hold)' in opened and 'offload/launch/window.sh open' not in opened |
| assert 'first stop refuses unowned/changed HOLD' in stop and 'guarded_stop' in stop |
| assert labels[-3:] == ['absent-fake-restore-close-x86host', 'commit-close', 'cancel-close'] |
| for phase in backend.phases: |
| fences = [labels.index('fence-' + phase + '-' + host) for host in w.HOSTS] |
| reaps = [labels.index('reap-' + phase + '-' + host) for host in w.HOSTS] |
| absent = [labels.index('absent-' + phase + '-' + host) for host in w.HOSTS] |
| assert max(fences) < min(reaps) and max(reaps) < min(absent) |
| for label, argv, timeout in transport.calls: |
| assert timeout <= 1800 |
| if label.startswith('run-'): |
| assert 'launch' in argv[-1] and 'SCOPED_REMOTE_DEADLINE=2795.000000' in argv[-1] |
| with pytest.raises(w.WatchError, match='already admitted'): |
| backend.recover(1900) |
|
|
|
|
| @pytest.mark.parametrize('failure', ['run-hold-open', 'run-stop', 'quiet-sparkhead', 'quiet-sparkworker-wired', |
| 'run-restore-start', 'run-restore-check', |
| 'fence-fake-restore-close-x86host', 'absent-fake-restore-close-x86host']) |
| def test_failures_retain_hold_and_never_commit(recovery: tuple[w.ScopedRecovery, Transport], failure: str) -> None: |
| backend, transport = recovery |
| transport.failure = failure |
| with pytest.raises(w.WatchError): |
| backend.recover(1800) |
| assert 'commit-close' not in [label for label, _, _ in transport.calls] |
|
|
|
|
| def test_fence_failure_prevents_every_reap_but_fences_all_hosts(recovery: tuple[w.ScopedRecovery, Transport]) -> None: |
| backend, transport = recovery |
| transport.failure = 'fence-fake-hold-open-sparkhead' |
| with pytest.raises(w.WatchError): |
| backend.recover(1800) |
| labels = [label for label, _, _ in transport.calls] |
| assert 'fence-fake-hold-open-x86host' in labels |
| assert not any(label.startswith('reap-') for label in labels) |
|
|
|
|
| def test_lost_commit_reply_cancels_and_retains_original_hold(recovery: tuple[w.ScopedRecovery, Transport]) -> None: |
| backend, transport = recovery |
| transport.failure = 'commit-close' |
| with pytest.raises(w.WatchError, match='unconfirmed'): |
| backend.recover(1800) |
| assert [label for label, _, _ in transport.calls][-3:] == ['commit-close', 'cancel-close', 'retain-close'] |
|
|
|
|
| def test_expired_lease_never_calls_transport(recovery: tuple[w.ScopedRecovery, Transport]) -> None: |
| backend, transport = recovery |
| backend.deadline = -1 |
| with pytest.raises(w.WatchError, match='deadline'): |
| backend.invoke('late', 'sparkhead', 'false') |
| assert not transport.calls |
|
|
|
|
| def test_dry_default_has_no_network_or_files(capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch) -> None: |
| def forbidden(*args: object, **kwargs: object) -> object: |
| raise AssertionError('dry plan performed I/O') |
| monkeypatch.setattr(urllib.request, 'urlopen', forbidden) |
| monkeypatch.setattr(subprocess, 'Popen', forbidden) |
| monkeypatch.setattr(Path, 'mkdir', forbidden) |
| assert w.main([]) == 0 |
| assert 'PLAN ONLY' in capsys.readouterr().out |
|
|
|
|
| def test_recovery_restores_signal_handlers_on_failure() -> None: |
| signals = (signal.SIGINT, signal.SIGTERM, signal.SIGHUP) |
| before = {sig: signal.getsignal(sig) for sig in signals} |
| with pytest.raises(RuntimeError): |
| with w.masked_signals(): |
| assert all(signal.getsignal(sig) == signal.SIG_IGN for sig in signals) |
| raise RuntimeError('fake failure') |
| assert {sig: signal.getsignal(sig) for sig in signals} == before |
|
|
|
|
| def test_w8_pin_drift_refuses_before_any_mutation(monkeypatch: pytest.MonkeyPatch) -> None: |
| transport = Transport() |
| monkeypatch.setattr(Path, 'read_bytes', lambda _: b'changed W8') |
| backend = w.ScopedRecovery('fake', Path('/unused'), call=transport, clock=lambda: 0, wall=lambda: 1000) |
| with pytest.raises(w.WatchError, match='changed'): |
| backend.recover(1800) |
| assert not transport.calls |
|
|
|
|
| def test_deadline_crossing_cannot_advance_to_next_phase(recovery: tuple[w.ScopedRecovery, Transport]) -> None: |
| backend, transport = recovery |
| clock = Clock() |
| backend.clock = clock |
| original = transport.__call__ |
| def call(label: str, argv: list[str], timeout: float) -> tuple[int, str]: |
| result = original(label, argv, timeout) |
| if label == 'run-stop': |
| clock.value = 1801 |
| return result |
| backend.call = call |
| with pytest.raises(w.WatchError, match='deadline|absence unproved'): |
| backend.recover(1800) |
| labels = [label for label, _, _ in transport.calls] |
| assert 'run-stop' in labels |
| assert 'run-restore-start' not in labels and 'commit-close' not in labels |
| assert backend.deadline == 1800 |
|
|
|
|
| def test_ambiguous_launch_is_fenced_on_every_destination(recovery: tuple[w.ScopedRecovery, Transport]) -> None: |
| backend, transport = recovery |
| original = transport.__call__ |
| def call(label: str, argv: list[str], timeout: float) -> tuple[int, str]: |
| if label == 'run-hold-open': |
| transport.calls.append((label, argv, timeout)) |
| raise OSError('fake reply lost after remote HOLD write') |
| return original(label, argv, timeout) |
| backend.call = call |
| with pytest.raises(OSError, match='reply lost'): |
| backend.recover(1800) |
| labels = [label for label, _, _ in transport.calls] |
| for host in w.HOSTS: |
| assert 'fence-fake-hold-open-' + host in labels |
| assert 'absent-fake-hold-open-' + host in labels |
| assert not any(label.startswith('run-restore-') for label in labels) |
|
|
|
|
| def test_execute_without_coordinator_refuses_before_any_write_or_probe(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| monkeypatch.delenv('WINDOW_COORDINATOR', raising=False) |
| out = tmp_path/'watch' |
| with pytest.raises(w.WatchError, match='WINDOW_COORDINATOR'): |
| w.main(['--execute', '--tag', 'fixture', '--out', str(out)]) |
| assert not out.exists() |
|
|
|
|
| def test_late_cancel_reply_still_retains_hold_after_deadline(recovery: tuple[w.ScopedRecovery, Transport]) -> None: |
| backend, transport = recovery |
| clock = Clock() |
| backend.clock = clock |
| original = transport.__call__ |
| def late(label: str, argv: list[str], timeout: float) -> tuple[int, str]: |
| result = original(label, argv, timeout) |
| if label == 'cancel-close': |
| clock.value = 1801 |
| return result |
| backend.call = late |
| with pytest.raises(w.WatchError, match='after immutable deadline'): |
| backend.recover(1800) |
| assert [label for label, _, _ in transport.calls][-3:] == ['commit-close', 'cancel-close', 'retain-close'] |
| assert backend.deadline == 1800 |
|
|
|
|
| def test_default_logger_survives_repeated_phase_labels(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: |
| |
| monkeypatch.setattr(Path, 'read_bytes', lambda _: b'reviewed W8') |
| monkeypatch.setattr(w, 'RESTORE_SHA256', hashlib.sha256(b'reviewed W8').hexdigest()) |
| transport = Transport() |
| backend = w.ScopedRecovery('fake', tmp_path / 'commands', clock=lambda: 0, wall=lambda: 1000) |
| def logged(label: str, argv: list[str], timeout: float) -> tuple[int, str]: |
| rc, text = transport(label, argv, timeout) |
| return backend.local(label, ['bash', '-c', f'printf %s {shlex.quote(text)}; exit {rc}'], timeout) |
| backend.call = logged |
| backend.recover(1800) |
| labels = [label for label, _, _ in transport.calls] |
| assert len(set(labels)) < len(labels) |
| logs = sorted((tmp_path / 'commands').iterdir()) |
| assert [p.name for p in logs] == [f'{n:04d}-{label}.log' for n, label in enumerate(labels, 1)] |
|
|
|
|
