| """PP4 shared-owner integration; CPU fixtures only, real scopes for receiving hops.""" |
| from __future__ import annotations |
|
|
| import json |
| import shlex |
| import subprocess |
| from pathlib import Path |
| from typing import cast |
|
|
| import pytest |
|
|
| from offload.launch.windows import hold_admission |
| from offload.launch.windows import pf1_window as pf |
| from offload.launch.windows import pp4_window as w |
| from offload.launch.windows import scoped_remote as owner |
| from offload.launch.windows.tests.test_pp4_window import WORKER |
| from offload.launch.windows.tests.test_scoped_remote import Sandbox, wait_for |
|
|
| pytest_plugins = ["offload.launch.windows.tests.test_scoped_remote"] |
|
|
|
|
| @pytest.mark.parametrize('kind', ['stop', 'run']) |
| @pytest.mark.parametrize('pause', ['before-registration', 'active']) |
| def test_real_pp4_owned_second_hop_and_retry(sandbox: Sandbox, kind: str, pause: str, monkeypatch: pytest.MonkeyPatch) -> None: |
| s = sandbox |
| tag = s.tag('pp4-' + kind) |
| home = s.homes['sparkworker-wired'] |
| state = home/'production' |
| state.write_text('active') |
| release = home/'release' |
| if pause == 'before-registration': |
| (s.root/'sparkworker-wired-transport-gate').touch() |
| body = f'touch {home}/active; while [ ! -e {release} ]; do sleep .02; done; echo {kind} > {state}' |
| live = w.Live(s.root, pf.Supervisor(s.root), native=WORKER, remote='experiment', expected=WORKER.commit, |
| data_dir=s.root, tag=tag, admission={'verifier_sha256': {}}) |
| live.sup.abort_at = s.deadlines[tag] |
| submitted: list[str] = [] |
| def ssh(label: str, host: str, command: str, stdin: str | None = None, *, timeout: float = 120) -> str: |
| submitted.append(command) |
| return '' |
| monkeypatch.setattr(live, 'ssh', ssh) |
| live.owned('engine-start', 'sparkhead', tag, 'ssh sparkworker ' + shlex.quote(body), timeout=30) |
| assert live.owner_tags == [tag] and live.owner_deadlines[tag] == s.deadlines[tag] |
| process = subprocess.Popen(['bash', '-c', submitted[0]], env=s.env(), stdout=subprocess.DEVNULL, |
| stderr=subprocess.DEVNULL, text=True) |
| s.processes.append(process) |
| wait_for(s.root/'sparkworker-wired-accepted' if pause == 'before-registration' else home/'active') |
| s.close('new', tag) |
| (s.root/'sparkworker-wired-transport-gate').unlink(missing_ok=True) |
| release.touch() |
| process.wait(timeout=10) |
| assert state.read_text() == 'active' |
| retry = s.run(s.command('new', 'launch', tag, ['bash', '-c', body]), 'sparkworker-wired') |
| assert retry.returncode == 75 and state.read_text() == 'active' |
| live.sup.close() |
|
|
|
|
| def test_s55_lease_origin_admission_and_no_renewal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| clock = [1000.] |
| sup = pf.Supervisor(tmp_path, clock=lambda: clock[0]) |
| live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit, |
| data_dir=tmp_path, tag='lease', admission={'verifier_sha256': {}}, clock=lambda: clock[0]) |
| clock[0] += 1080 |
| before = clock[0] |
| calls: list[str] = [] |
| def ssh(label: str, host: str, command: str, stdin: str | None = None, *, timeout: float = 120) -> str: |
| calls.append(label) |
| if label == 'hold-open': |
| assert 'SCOPED_REMOTE_DEADLINE=' in command and hold_admission.NOT_OWNED in command |
| clock[0] += pf.DRAIN_S - 10 |
| from offload.launch.windows.tests.test_pf1_window import iso |
| return iso(clock[0]) + ' DSV41-OFFLOAD (session coordinator): PP4-WINDOW lease: fixture' |
| monkeypatch.setattr(live, 'ssh', ssh) |
| monkeypatch.setattr(sup, 'tunnel', lambda argv: None) |
| live.open_hold() |
| assert live.open_deadline == before + pf.DRAIN_S + 60 |
| assert live.owner_deadlines['lease-hold'] == live.hold_scope_deadline >= live.restore_by |
| saved = live.restore_by, live.hold_scope_deadline |
| with pytest.raises(ValueError, match='cannot be reopened or renewed'): |
| live.open_hold() |
| assert saved == (live.restore_by, live.hold_scope_deadline) |
| sup.close() |
|
|
|
|
| @pytest.mark.parametrize('mode', ['fence', 'reap', 'absent']) |
| @pytest.mark.parametrize('host', w.OWNER_HOSTS) |
| def test_all_host_control_failures_prevent_pair_and_w8(tmp_path: Path, mode: str, host: str) -> None: |
| calls: list[str] = [] |
| def plain(label: str, argv: list[str], timeout: float) -> tuple[int, str]: |
| calls.append(label) |
| if label == f'owned-{mode}-engine-{host}': |
| return 1, 'injected missing proof' |
| return 0, owner.FENCED if label.startswith('owned-fence-') else owner.ABSENT |
| sup = pf.Supervisor(tmp_path) |
| live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit, |
| data_dir=tmp_path, tag='fixture', admission={'verifier_sha256': {}}, plain=plain) |
| live.hold_written = True |
| live.remember('engine', live.restore_by - 90) |
| with pytest.raises(ValueError, match='HOLD stays open'): |
| live.restore(production_stopped=True) |
| assert 'pair-close' not in calls and 'start' not in calls and 'commit-close' not in calls |
| if mode == 'fence': |
| assert not any(label.startswith(('owned-reap-', 'owned-absent-')) for label in calls) |
|
|
|
|
| def test_pins_only_shared_ownership_and_recovery_is_scoped() -> None: |
| assert 'offload/launch/windows/scoped_remote.py' in w.PIN_FILES |
| assert not (pf.ROOT/'offload/launch/windows/pp4_remote.py').exists() |
| assert {'offload/launch/windows/'+name+'.py' for name in |
| ('post_close_watch', 'hold_admission', 'spark_quiet', 'production_health')} <= set(w.PIN_FILES) |
|
|
|
|
| def test_repeated_recovery_proofs_keep_distinct_logs(tmp_path: Path) -> None: |
| sup = pf.Supervisor(tmp_path) |
| (tmp_path/'cmd').mkdir(exist_ok=True) |
| live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit, |
| data_dir=tmp_path, tag='repeat', admission={'verifier_sha256': {}}) |
| for text in ('first proof', 'second proof'): |
| assert live.local('same-phase-absent', ['bash', '-c', 'printf %s '+shlex.quote(text)], 5) == (0, text) |
| logs = sorted((tmp_path/'cmd').glob('restore-*.log')) |
| assert [p.name for p in logs] == ['restore-0001-same-phase-absent.log', 'restore-0002-same-phase-absent.log'] |
| assert not (tmp_path/'cmd'/'restore-same-phase-absent.log').exists() |
| assert [p.read_text() for p in logs] == ['first proof', 'second proof'] |
| sup.close() |
|
|
|
|
| @pytest.mark.parametrize('restore,watch_status,exit_code', [ |
| ('RESTORED: verified', 'PASS', 0), ('RESTORED: verified', 'FAILED', 2), |
| ('FAILED', 'PASS', 2), ('NOT NEEDED: preflight only', 'PASS', 0), |
| ]) |
| def test_main_watchdog_is_mandatory_after_restore(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, |
| restore: str, watch_status: str, exit_code: int) -> None: |
| from types import SimpleNamespace |
|
|
| from offload.launch.windows import post_close_watch as watch |
| from offload.launch.windows import pp4_prerequisites as prereq |
| from offload.prefill_maxlen.evidence import Json, record |
|
|
| monkeypatch.setenv('WINDOW_COORDINATOR', 'Deepseek/coordinator') |
| monkeypatch.setattr(pf, 'CAMPAIGN', tmp_path) |
| monkeypatch.setattr(subprocess, 'check_output', lambda argv, **kwargs: WORKER.commit if argv[1] == 'rev-parse' else '') |
| monkeypatch.setattr(prereq, 'staging', lambda *args: {'worker_identity': WORKER.identity, 'markov_identity': WORKER.markov}) |
| calls: list[str] = [] |
| monkeypatch.setattr(pf, 'Supervisor', lambda out: SimpleNamespace(close=lambda: calls.append('close'))) |
| monkeypatch.setattr(w, 'Live', lambda *args, **kwargs: SimpleNamespace()) |
| monkeypatch.setattr(w, 'window', lambda *args, **kwargs: { |
| 'status': 'PAIR_QUALIFIED', 'restore': {'status': restore}, 'decision': {'status': 'CANDIDATE'}}) |
| monkeypatch.setattr(watch, 'ProductionHealth', object) |
| def recovery(tag: str, out: Path) -> object: |
| assert calls == ['close'] |
| assert tag == 'PP4-test-watch' |
| calls.append('recovery') |
| return object() |
| def watchdog(health: object, recovery: object, *, log: object) -> watch.Result: |
| calls.append('watch') |
| return watch.Result(watch_status, 30, 0, 0) |
| monkeypatch.setattr(watch, 'ScopedRecovery', recovery) |
| monkeypatch.setattr(watch, 'watch', watchdog) |
| import signal |
| saved = {sig: signal.getsignal(sig) for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP)} |
| try: |
| rc = w.main(['--execute', '--pair-only', '--qualification', str(tmp_path/'staging.json'), |
| '--expected-experiment-commit', WORKER.commit, '--worker-commit', WORKER.commit, |
| '--worker-tree', WORKER.tree, '--worker-identity', WORKER.identity, |
| '--worker-markov-identity', WORKER.markov, '--out', str(tmp_path/'PP4-test')]) |
| finally: |
| for sig, handler in saved.items(): |
| signal.signal(sig, handler) |
| assert rc == exit_code |
| assert ('watch' in calls) == restore.startswith('RESTORED') |
| result = record(cast(Json, json.loads((tmp_path/'PP4-test/summary.json').read_text()))) |
| if restore.startswith('RESTORED'): |
| assert record(result['post_close_watch'])['status'] == watch_status |
| assert record(result['decision'])['status'] == ('CANDIDATE' if watch_status == 'PASS' else 'REJECT') |
|
|
|
|
| def test_local_control_precheck_never_requires_loopback_ssh(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| sup = pf.Supervisor(tmp_path) |
| live = w.Live(tmp_path, sup, native=WORKER, remote='experiment', expected=WORKER.commit, |
| data_dir=tmp_path, tag='fresh', admission={'verifier_sha256': {}}) |
| calls: list[list[str]] = [] |
| def call(label: str, argv: list[str], **kwargs: object) -> str: |
| calls.append(argv) |
| return '' |
| monkeypatch.setattr(live, 'call', call) |
| live.fresh_controls(['fresh-hold', 'fresh-worker']) |
| assert [argv[-2] for argv in calls if argv[0] == 'ssh'] == list(owner.NODES) |
| local = [argv for argv in calls if argv[0] != 'ssh'] |
| assert len(local) == 1 and local[0][:2] == ['bash', '-c'] |
| assert 'scoped-control/fresh-hold' in local[0][-1] and 'scoped-control/fresh-worker' in local[0][-1] |
| sup.close() |
|
|
