Blame for code/offload/launch/windows/up2_low.py

"""Direct W8-style UP2 low-cap runner. No window opening/closing or restoration."""
from __future__ import annotations

import argparse
import json
import os
import re
import selectors
import shlex
import signal
import subprocess
import sys
import time
from collections.abc import Callable
from pathlib import Path
from typing import IO, Protocol, cast

from offload.prefill_maxlen.evidence import Json, number, record
from offload.launch.windows import up2_checks as ck

CAMPAIGN = Path('/opt/dsv41/campaigns/DSV41-OFFLOAD')
ROOT = Path(__file__).resolve().parents[3]
RESTORE = ('# Inspect rollback.json; drain/cancel only window-owned requests before restoration.\n'
    'ssh sparkhead \'cd ~/dsv41-deploy && test "$(git rev-parse --short=7 HEAD)" = 21bed31 '
    '&& git diff --exit-code HEAD -- dsv41.env serve.sh start.sh\'\n' +
    '\n'.join(f'bash {CAMPAIGN}/W8/restore.sh {step}' for step in ('start','check','close')))
X86HOST_PROBE = '''import json,pathlib,os
root=pathlib.Path.home()/'dsv41-offload/prod'
current=(root/'dspark-worker/current').resolve()
status=json.loads((root/'run/dspark-worker.status.json').read_text())
process=pathlib.Path('/proc')/str(status['pid'])
env=dict(x.split('=',1) for x in (process/'environ').read_text().split('\\0') if '=' in x)
print(json.dumps({'revision':(current/'SOURCE_COMMIT').read_text().strip(),'current':str(current),
 'cwd':os.readlink(process/'cwd'),'cmd':(process/'cmdline').read_text().split('\\0'),
 'gpus':env.get('CUDA_VISIBLE_DEVICES'),'status':status}))
'''
ENGRAM_PROBE = '''import hashlib,json,os,pathlib,subprocess
root=pathlib.Path(os.environ['VLLM_ENGRAM_REMOTE_ROOT'])
files=[pathlib.Path(os.environ['DSV41_ENGRAM_OVERLAY']),root/'manifest.json',
 root/'offload/engram/spark/py/sitecustomize.py',
 pathlib.Path('/usr/local/lib/python3.12/dist-packages/vllm/models/deepseek_v41/common/engram_disk.cpp'),
 pathlib.Path('/opt/dsv41-ofl/libofl.so')]
files += [root/p for p in ('offload/engram/spark/engram_remote.cpp',
 'offload/engram/engram_proto.cpp','offload/engram/engram_client.h',
 'offload/engram/engram_proto.h','offload/transport/ofl.cpp','offload/transport/ofl.h')]
jit=set()
for proc in pathlib.Path('/proc').iterdir():
 if not proc.name.isdigit(): continue
 try: maps=(proc/'maps').read_text()
 except (FileNotFoundError,ProcessLookupError): continue
 for line in maps.splitlines():
  parts=line.split(None,5)
  if len(parts)==6 and '/vllm_engram_remote_' in parts[5] and parts[5].endswith('.so'):
   jit.add(pathlib.Path(parts[5]))
if len(jit)!=1: raise SystemExit('required single loaded Engram JIT identity missing/ambiguous')
files += sorted(jit)
hashes={}
for path in files:
 p=subprocess.Popen(['dd','if='+str(path),'iflag=direct','bs=4096','status=none'],stdout=subprocess.PIPE)
 try: hashes[str(path)]=hashlib.file_digest(p.stdout,'sha256').hexdigest()
 finally: p.stdout.close(); code=p.wait()
 if code: raise SystemExit('Engram O_DIRECT identity read failed: '+str(path))
print(json.dumps(hashes,sort_keys=True))
'''
ARMS = [('diag-off',0,1), ('diag-on',1,1), ('off-a',0,0), ('on',1,0), ('off-b',0,0)]
SSH = ['ssh','-o','BatchMode=yes','-o','ConnectTimeout=4','-o','ServerAliveInterval=4','-o','ServerAliveCountMax=1']


def save(path: Path, data: dict[str, Json], *, screen: bool = False) -> None:
    with path.open('x') as output:
        json.dump(ck.screen_record(data,screen), output, indent=2, allow_nan=False)
        output.write('\n')


class Backend(Protocol):
    out: Path
    def start(self, arm: str, replay: int, audit: int) -> None: ...
    def request(self, arm: str, name: str, target: int, budget: int, *, numeric: bool = False,
                force_length: bool = False, rep: int = 0, cold: bool = False) -> dict[str, Json]: ...
    def native(self, arm: str, *, complete: bool) -> int: ...
    def needles(self, arm: str) -> None: ...
    def clean(self, arm: str, audit: int) -> None: ...


def sequence(backend: Backend, phase: str, checkpoint: Callable[[dict[str, Json]], None],
             *, screen: bool = False, contexts: tuple[int, ...] = (150000, 300000)) -> dict[str, Json]:
    """Same orchestration for real execution and offline rehearsal; every cell is durable."""
    def emit(data: dict[str, Json]) -> None:
        checkpoint(ck.screen_record(data,screen))
    ck.measured_contexts(contexts)
    diagnostic_status = 'PASS-SCREEN' if screen else 'PASS'
    cells: dict[str, list[dict[str, Json]]] = {}
    numeric: dict[str, list[dict[str, Json]]] = {}
    if phase in ('all','diagnostics'):
        for arm,replay,audit in ARMS[:2]:
            backend.start(arm,replay,audit)
            backend.request(arm,'warm32',32000,32,force_length=True)
            q = backend.native(arm,complete=False)
            for tail in (1,127,128,129):
                backend.request(arm,f'tail{tail}',36*q+tail,8,force_length=True)
            backend.request(arm,'near-cap',327675,5,force_length=True)
            numerical_rows: list[dict[str, Json]] = []
            for n in (512,32000,150000,300000):
                for rep in range(2 if arm == 'diag-off' else 1):
                    numerical_rows.append(backend.request(arm,f'num{n}-{rep}',n,64,numeric=True,rep=rep))
            numeric[arm] = numerical_rows
            backend.needles(arm)
            backend.native(arm,complete=True)
            backend.clean(arm,audit)
            if arm == 'diag-on':
                comparisons: list[Json] = []
                for i,n in enumerate((512,32000,150000,300000)):
                    a,aa = numeric['diag-off'][2*i:2*i+2]
                    b = numerical_rows[i]
                    ck.require(a['prompt_sha256'] == aa['prompt_sha256'] == b['prompt_sha256'], 'numeric prompt drift')
                    # Greedy text is a record at every context, A/A included: the engine is not run-to-run
                    # deterministic at T=0 (W13 r2: same-start OFF A/A text differed at 512, 32K, 150K and 300K with
                    # first-token KL 0.0 at 512). The numerical floor above is the hard gate.
                    comparisons.append({'context':n, **ck.numerical(a,aa,b),
                        'text_aa_equal_INFORMATIONAL': a['text_sha256'] == aa['text_sha256'] and a['finish'] == aa['finish'],
                        'text_on_equals_off_INFORMATIONAL': a['text_sha256'] == b['text_sha256'] and a['finish'] == b['finish']})
                emit({'numerical':comparisons,'diagnostics':diagnostic_status})
    if phase in ('all','measured'):
        for arm,replay,audit in ARMS[2:]:
            backend.start(arm,replay,audit)
            backend.request(arm,'warm32',32000,32,force_length=True)
            cells[arm] = []
            for n in contexts:
                for rep in range(3):
                    row = ck.screen_record(backend.request(arm,f'cold{n}-{rep}',n,1500,rep=rep,cold=True),screen)
                    ck.require(row.get("window_id") == str(backend.out.resolve()), "cell belongs to another window")
                    ck.cold(row,n,rep)
                    cells[arm].append(row)
                    emit(ck.summarize(cells,contexts))
            backend.clean(arm,audit)
    result = {'phase':phase, 'diagnostics':diagnostic_status if numeric else 'PRIOR QUALIFICATION REQUIRED',
              **ck.summarize(cells,contexts)}
    if screen and numeric:
        result['native'] = 'NOT RUN' if getattr(backend,'native_missing',False) else 'PASS'
    return ck.screen_record(result,screen)


def terminate(proc: subprocess.Popen[str]) -> None:
    if proc.poll() is None:
        os.killpg(proc.pid,signal.SIGTERM)
        try:
            proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            os.killpg(proc.pid,signal.SIGKILL)
            proc.wait(timeout=3)


class Supervisor:
    """Only owned local process groups. Continuous5s remote read-only memory streams."""
    def __init__(self, out: Path, deadline: float):
        self.out, self.deadline = out, deadline
        self.select = selectors.DefaultSelector()
        self.monitors: list[subprocess.Popen[str]] = []
        self.files: list[IO[str]] = []
        self.pending: dict[int, str] = {}
        self.last: dict[str, float] = {}
        self.active: subprocess.Popen[str] | None = None

    def begin(self) -> None:
        script = ('import time; from pathlib import Path\nwhile True:\n '
            'print(";".join(x for x in Path("/proc/meminfo").read_text().splitlines() '
            'if x.startswith(("MemAvailable:","SwapFree:"))),flush=True); time.sleep(5)')
        for host in ('sparkhead','sparkworker-wired'):
            err = (self.out/f'memory-{host}.stderr').open('x')
            log = (self.out/f'memory-{host}.jsonl').open('x')
            self.files.extend([err,log])
            proc = subprocess.Popen([*SSH,host,'exec python3 -u -c '+shlex.quote(script)],
                stdout=subprocess.PIPE,stderr=err,text=True,start_new_session=True)
            self.monitors.append(proc)
            assert proc.stdout is not None
            os.set_blocking(proc.stdout.fileno(),False)
            self.select.register(proc.stdout,selectors.EVENT_READ,(host,log))
            self.pending[proc.stdout.fileno()] = ''
        start = time.monotonic()
        while len(self.last) < 2:
            ck.require(time.monotonic()-start < 10, 'initial memory telemetry missing')
            self.poll()

    def poll(self) -> None:
        ck.require(not (self.out/'STOP').exists(), 'external STOP file')
        ck.require(time.time() < self.deadline, f'90-minute hard deadline {self.deadline:.6f} reached')
        for key,_ in self.select.select(timeout=.1):
            stream = cast(IO[str],key.fileobj)
            part = os.read(stream.fileno(),65536).decode()
            ck.require(bool(part), 'memory monitor EOF')
            self.pending[stream.fileno()] += part
            while '\n' in self.pending[stream.fileno()]:
                line,self.pending[stream.fileno()] = self.pending[stream.fileno()].split('\n',1)
                host,log = key.data
                available,swap = ck.memory(line.replace(';','\n'))
                self.last[host] = time.monotonic()
                log.write(json.dumps({'time':time.time(),'available_kb':available,'swap_free_kb':swap})+'\n')
                log.flush()
        ck.require(all(p.poll() is None for p in self.monitors), 'memory monitor exited')
        ck.require(all(time.monotonic()-t < 12 for t in self.last.values()), 'stale memory telemetry')

    def run(self, name: str, command: list[str], *, stdin: str | None = None, timeout: float = 900) -> str:
        self.poll()
        start = time.monotonic()
        with (self.out/f'{name}.log').open('x') as log:
            # File stdin avoids a pipe write deadlock for the emitted42-file verifier.
            input_path = self.out/f'{name}.stdin'
            if stdin is not None:
                input_path.write_text(stdin)
            with input_path.open() if stdin is not None else open(os.devnull) as source:
                proc = subprocess.Popen(command,stdin=source,stdout=log,stderr=subprocess.STDOUT,
                                        text=True,start_new_session=True)
                self.active = proc
                try:
                    while proc.poll() is None:
                        self.poll()
                        ck.require(time.monotonic()-start < timeout, f'{name} timeout/nonprogress')
                    if proc.returncode != 0:
                        log.flush()
                        # Preserve the request driver's exact deadline refusal, rather
                        # than guessing from the clock after an unrelated child error.
                        deadline_error = re.search(
                            r'^ValueError: STOP: ((?:needle )?submission deadline [0-9.]+ reached)$',
                            (self.out/f'{name}.log').read_text(), re.M)
                        if deadline_error:
                            ck.require(False, f'{name}: {deadline_error.group(1)}')
                    ck.require(proc.returncode == 0,f'{name} failed rc{proc.returncode}; see private log')
                    self.poll()
                finally:
                    terminate(proc)
                    self.active = None
        return (self.out/f'{name}.log').read_text()

    def close(self) -> None:
        if self.active is not None:
            terminate(self.active)
        for proc in self.monitors:
            terminate(proc)
        self.select.close()
        for handle in self.files:
            handle.close()


class Live:
    screen = False
    native_missing = False

    def __init__(self, out: Path, remote: str, expected: str, deadline: float, submit: float,
                 native_root: Path, data_dir: Path, supervisor: Supervisor, native_wait: float = 0, *, screen: bool = False):
        self.out,self.remote,self.expected = out,remote,expected
        self.submit,self.native_root,self.data_dir = submit,native_root,data_dir
        self.s = supervisor
        self.native_wait = native_wait
        self.screen = screen
        self.tail_q: dict[str,int] = {}
        self.serial = 0
        self.production: list[dict[str, Json]] = []
        self.ids: list[str] = []
        self.previous_ids: set[str] = set()
        self.before: dict[str, Json] | None = None
        self.last_stats: dict[str, Json] = {}
        self.proxy_pid = 0
        self.manifest = ck.digest(ROOT/'offload/upstream_adopt/MANIFEST.json')
        self.verifier = self.s.run('emit-verifier',[sys.executable,'offload/upstream_adopt/verify_runtime.py','--emit'])

    def save(self, path: Path, data: dict[str, Json]) -> None:
        save(path,data,screen=self.screen)

    def call(self, label: str, command: list[str], stdin: str | None = None) -> str:
        self.serial += 1
        return self.s.run(f'{self.serial:04d}-{label}',command,stdin=stdin)

    def ssh(self, host: str, command: str, stdin: str | None = None) -> str:
        return self.call(host,[*SSH,host,command],stdin)

    def health(self, *, production: bool = False) -> None:
        port = 8000 if production else 8001
        self.ssh('sparkhead',f'curl -fsS --max-time 10 http://localhost:{port}/health')
        info = record(cast(Json,json.loads(self.ssh('sparkhead',f'curl -fsS --max-time 10 http://localhost:{port}/v1/models'))))
        models = ck.rows(info['data'])
        ck.require(len(models) == 1 and models[0].get('max_model_len') == ck.CAP and
                   models[0].get('root') == '/srv/dsv41/legal-v2a', 'model identity/cap failure')

    def idle(self, *, production: bool = False) -> None:
        port = 8000 if production else 8001
        metrics = self.ssh('sparkhead',f'curl -fsS --max-time 10 http://localhost:{port}/metrics')
        for key in ('vllm:num_requests_running','vllm:num_requests_waiting'):
            found = [float(x.split()[-1]) for x in metrics.splitlines() if x.startswith((key+'{',key+' '))]
            ck.require(bool(found) and sum(found) == 0,f'{key} absent/busy')

    def worker(self) -> None:
        text = self.ssh('x86host','python3 -c '+shlex.quote(X86HOST_PROBE))
        probe = record(cast(Json,json.loads(text)))
        ck.require(probe.get('revision') == ck.WORKER,'worker source revision mismatch; never overwrite')
        ck.require(str(probe['cwd']).startswith(str(probe['current'])+'/'), 'running worker source differs from current')
        argv = ck.strings(probe['cmd'])
        ck.require(argv[argv.index('--tp')+1] == '2' and argv[argv.index('--max-model-len')+1] == '1048576',
                   'running worker layout/position support mismatch')
        ck.require(probe.get('gpus') == 'GPU-c4e7a371-d8a8-ebf0-918c-da3e38c87a93,GPU-7744b44b-8a13-b06d-a67e-f958fd74b8fc',
                   'running worker GPU identity mismatch')
        data = record(probe['status'])
        ck.require(data.get('tp') == 2 and data.get('tp_comm_errors') == 0 and
                   data.get('listeners') == ['10.202.28.1:18620'], 'worker status identity/error failure')
        import datetime
        stamp = datetime.datetime.fromisoformat(str(data['time']).replace('Z','+00:00')).timestamp()
        ck.require(abs(time.time()-stamp) < 30,'worker status stale')

    def capture(self, host: str) -> dict[str, Json]:
        text = self.ssh(host,'bash -s', (ROOT/'offload/dspark/d2/capture_window_identity.sh').read_text())
        objects = [record(cast(Json,json.loads(line))) for line in text.splitlines()]
        ck.require(len(objects) == 2,'incomplete rank identity capture')
        container,files = objects
        inspect = record(cast(Json,json.loads(self.ssh(host,"docker inspect --format '{{json .Config}}' dsv41-serve"))))
        env: dict[str, Json] = {}
        for entry in ck.strings(inspect['Env']):
            key,_,value = entry.partition('=')
            if key.startswith(('VLLM_','EXL3_','NCCL_','PYTORCH_','CUDA_')) or key in ('PYTHONPATH','DSV41_ENGRAM_OVERLAY'):
                ck.require(key not in env,'duplicate runtime environment key')
                env[key] = value
        # Only target metadata belongs in comparison: composed engine files intentionally differ.
        target = {k:v for k,v in record(files['files']).items() if k.startswith('/srv/dsv41/legal-v2a/')}
        target['generation_config'] = files['generation_config']
        return {'id':container['id'],'state':container['state'],'image':container['image'],
                'engine':files['engine'],'files':target,'env':env,
                'engram_identity':record(cast(Json,json.loads(self.ssh(host,
                    'docker exec -i dsv41-serve python3 -',ENGRAM_PROBE)))),
                'argv':self.ssh(host,'docker exec dsv41-serve cat /serve.sh')}

    def preflight(self) -> None:
        ck.require((CAMPAIGN/'D2-GEN.PAUSE').exists() or subprocess.run(
            ['tmux','has-session','-t','=d2-gen'],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL).returncode != 0,
            'D2 generation active without pause')
        self.ssh('sparkhead',"test -f ~/dsv41-deploy/HOLD && cd ~/dsv41-deploy && "
            "test \"$(git rev-parse --short=7 HEAD)\" = 21bed31 && "
            "test \"$(git rev-parse HEAD)\" = \"$(git rev-list -n1 prod-k1-327-20261023)\" && "
            "git diff --exit-code HEAD -- dsv41.env serve.sh start.sh")
        self.ssh('sparkhead',f'cd ~/{shlex.quote(self.remote)} && test "$(git rev-parse HEAD)" = {shlex.quote(self.expected)} '
            '&& test -z "$(git status --porcelain --untracked-files=no)"')
        # Source guard on CPU even for OFF. Staged replay guard is called again by start_exp for ON.
        from offload.upstream_adopt.replay_guard import verify
        config = self.call('source-config',['bash','-c',f'set -eu; source {ck.CONFIG}; printf "%s\\n%s\\n" "$OVERLAYS" "$EXTRA_DOCKER_ARGS"'])
        overlays,extra = config.splitlines()
        verify(overlays,extra)
        paths = [ck.CONFIG,'offload/launch/configs/UP2-0075.env',
                 'offload/launch/configs/D-kv6-async-up1-control.env','offload/launch/start_exp.sh',
                 'offload/launch/serve_exp.sh','offload/upstream_adopt/MANIFEST.json']
        remote = self.ssh('sparkhead',f'cd ~/{shlex.quote(self.remote)} && sha256sum '+shlex.join(paths))
        expected = '\n'.join(f'{ck.digest(ROOT/p)}  {p}' for p in paths)
        ck.require(remote.strip() == expected,'staged config/source mismatch')
        self.health(production=True)
        self.idle(production=True)
        self.worker()
        self.production = [self.capture(host) for host in ('sparkhead','sparkworker-wired')]
        ck.require(self.production[0]['files'] == self.production[1]['files'] and
                   self.production[0]['image'] == self.production[1]['image'], 'production rank identities disagree')
        self.previous_ids.update(str(r['id']) for r in self.production)
        self.save(self.out/'rollback.json',{'production_commit':'21bed31','tag':'prod-k1-327-20261023',
            'checkout':'~/dsv41-deploy','worker_source':ck.WORKER,'ranks':cast(Json,self.production)})

    def start(self, arm: str, replay: int, audit: int) -> None:
        ck.require(time.time() < self.submit,f'submission deadline {self.submit:.6f} reached; do not start another arm')
        self.ssh('sparkhead','test -f ~/dsv41-deploy/HOLD')
        self.idle(production=arm in ('diag-off','off-a') and not self.ids)
        self.worker()
        self.ssh('sparkhead',f'cd ~/{shlex.quote(self.remote)} && UP2_CAP=327680 UP2_REPLAY={replay} UP2_AUDIT={audit} '
            f'bash offload/launch/start_exp.sh {ck.CONFIG}')
        self.health()
        current = [self.capture(host) for host in ('sparkhead','sparkworker-wired')]
        for rank,host in enumerate(('sparkhead','sparkworker-wired')):
            ck.runtime(self.production[rank],current[rank],replay,audit)
            ck.require(str(current[rank]['id']) not in self.previous_ids,'start reused a previous container')
            result = self.ssh(host,'docker exec -i dsv41-serve python3 -',self.verifier)
            ck.require('UP1 runtime hashes verified: 42 files;' in result,'mounted42-file check missing')
            logs = self.ssh(host,'docker logs dsv41-serve 2>&1')
            ck.clean_logs(logs)
            ck.require('remote Engram' in logs or 'remote table' in logs or 'Engram remote' in logs,
                       'remote Engram runtime evidence missing')
            if rank == 0:
                # Only rank 0 runs EngineCore, which logs the KV capacity; rank 1 is --headless (workers only).
                found = re.findall(r'GPU KV cache size: ([\d,]+)',logs)
                ck.require(bool(found) and all(int(n.replace(',','')) == ck.KV for n in found),'rank KV capacity mismatch')
                pids = re.findall(r'through proxy process pid (\d+)',logs)
                ck.require(len(pids) == 1 and 'async mode on' in logs, 'async process startup not proven')
                self.proxy_pid = int(pids[0])
        self.ids = [str(r['id']) for r in current]
        self.previous_ids.update(self.ids)
        self.save(self.out/f'{arm}-identity.json',{'ranks':cast(Json,current)})
        self.before = None

    def stats(self) -> dict[str, Json]:
        self.idle()
        return record(cast(Json,json.loads(self.ssh('sparkhead',
            f'docker exec -i dsv41-serve python3 - --pid {self.proxy_pid}',
            (ROOT/'offload/launch/windows/up2_proxy_stats.py').read_text()))))

    def request(self, arm: str, name: str, target: int, budget: int, *, numeric: bool = False,
                force_length: bool = False, rep: int = 0, cold: bool = False) -> dict[str, Json]:
        self.health()
        self.idle()
        # Coordinator fault probes may deliberately exercise ordinary failures.
        # Normal requests must introduce none of the forbidden counter growth.
        before = self.stats()
        output = self.out/f'{arm}-{name}.json'
        # Pair numeric/cold requests across restarted arms. Warmup uses separate salts.
        salt = f'{self.out.name}:{"cold" if cold else name}:{target}:{rep}'
        cmd = [sys.executable,'-m','offload.launch.windows.up2_request','--target',str(target),'--budget',str(budget),
               '--salt',salt,'--output',str(output),'--ledger',str(self.out/'requests.jsonl'),
               '--deadline',str(self.submit),'--data-dir',str(self.data_dir),'--rep',str(rep),
               '--kind','cold' if cold else name]
        if self.screen: cmd.append('--screen')
        if numeric: cmd.append('--numeric')
        if force_length: cmd.append('--force-length')
        self.call(f'{arm}-{name}',cmd)
        result = ck.read(output)
        self.clean(arm,1 if arm.startswith('diag-') else 0, before)
        if cold or name == 'warm32':
            ck.require(number(record(self.last_stats['counters']).get('rounds_ok',0)) >
                       number(record(before['counters']).get('rounds_ok',0)), 'warmup/cold request had no remote drafting')
        return result

    def native(self, arm: str, *, complete: bool) -> int:
        path = self.native_root/arm/('qualification.json' if complete else 'q.json')
        stage = 'complete' if complete else 'q'
        self.save(self.out/f'{arm}-await-{stage}.json',{'arm':arm,'container_ids':cast(Json,self.ids),
            'manifest_sha256':self.manifest,'required_packet':str(path),'submission_deadline':self.submit,
            'native_checks':list(ck.NATIVE_CHECKS) if complete else ['native measured q'],
            'tail_q':self.tail_q.get(arm)})
        end = min(time.time()+self.native_wait,self.submit)
        while not path.is_file() and time.time() < end:
            self.s.poll()
        if not path.is_file():
            self.native_missing = True
            missing = self.out/f'{arm}-native-NOT-RUN.json'
            if not missing.exists():
                self.save(missing,{'status':'NOT RUN' if self.screen else 'NOT RUN/HOLD',
                    'missing':str(path),'checks':{key:'NOT RUN' for key in ck.NATIVE_CHECKS}})
            if not self.screen:
                raise ValueError('NOT RUN/HOLD: native probe evidence unavailable; see UP2-LOW.md')
            if not complete:
                # The reviewed proxy publishes audited-round counts, not per-chunk q.
                # A count or MAX_BATCH is not a measured solo q. Preserve the snapshot
                # so this ESTIMATED fallback can be audited without inventing telemetry.
                self.tail_q[arm] = 4092
                self.save(self.out/f'{arm}-tails.json',{'q':4092,'classification':'ESTIMATED',
                    'source':'CPU schedule; reviewed audit/proxy stats expose no per-chunk q',
                    'proxy_stats':self.last_stats,'tails_verified':False,
                    'tail_sizes':[36*4092+t for t in (1,127,128,129)]})
            return self.tail_q[arm]
        packet = ck.read(path)
        q = ck.native(packet,arm,self.ids,self.manifest,self.native_root,complete=complete)
        if complete:
            ck.require(q == self.tail_q.get(arm),'native receipt q differs from q used for sent tails')
        else:
            self.tail_q[arm] = q
            if self.screen:
                self.save(self.out/f'{arm}-tails.json',{'q':q,'classification':'MEASURED',
                    'source':str(path),'tails_verified':False,
                    'tail_sizes':[36*q+t for t in (1,127,128,129)]})
        self.save(self.out/f'{arm}-native-{"complete" if complete else "q"}.json',packet)
        if complete:
            # Expected ordinary fault failures are scoped by the native receipt.
            # They cannot be used as a baseline for normal requests until verified.
            self.before = self.stats()
        return q

    def needles(self, arm: str) -> None:
        for target in (150000,300000):
            self.idle()
            before = self.stats()
            path = self.out/f'{arm}-needle{target}.json'
            self.call(f'{arm}-needle{target}',[sys.executable,'-m','offload.launch.windows.up2_request',
                '--needle','--target',str(target),'--budget','300','--salt',f'{self.out.name}:needle{target}',
                '--output',str(path),'--ledger',str(self.out/'requests.jsonl'),'--deadline',str(self.submit),
                '--data-dir',str(self.data_dir),*(['--screen'] if self.screen else [])])
            data = ck.read(path)
            ck.require(data.get('recall') == 6 and data.get('of') == 6 and data.get('foreign_codes') == 0,
                       'needle6/6 or foreign-code failure')
            self.clean(arm,1,before)

    def clean(self, arm: str, audit: int, before: dict[str, Json] | None = None) -> None:
        self.health()
        data = self.stats()
        ck.counters(data,audit,before or self.before)
        with (self.out/'proxy-stats.jsonl').open('a') as handle:
            handle.write(json.dumps(ck.screen_record({'arm':arm,'time':time.time(),
                'counters':data['counters'],'async_watchdog':data['async_watchdog']},self.screen),allow_nan=False)+'\n')
        self.last_stats = data
        for rank,host in enumerate(('sparkhead','sparkworker-wired')):
            identity = self.ssh(host,"docker inspect --format '{{.Id}}' dsv41-serve").strip()
            ck.require(identity == self.ids[rank],'engine/container changed during arm')
            ck.clean_logs(self.ssh(host,'docker logs dsv41-serve 2>&1'))
        self.worker()
        self.idle()


def plan(phase: str, *, screen: bool = False, contexts: tuple[int, ...] = (150000, 300000)) -> str:
    ck.measured_contexts(contexts)
    selected = ARMS if phase == 'all' else ARMS[:2] if phase == 'diagnostics' else ARMS[2:]
    lines = ['DRY RUN: no commands executed; no network or output files.',
        'Coordinator Deepseek/coordinator; HOLD already open; current production snapshot + idle + worker + source guard.',
        'Continuous both-rank5s memory supervision >=3GiB; SwapFree recorded; STOP/deadline abort.',
        'Each start: idle, source/worker pins, start_exp, cap/KV, both-rank verify_runtime42 files, argv/env, proxy/Engram/no local drafter.']
    for arm,replay,audit in selected:
        lines.append(f'{arm}: UP2_CAP=327680 UP2_REPLAY={replay} UP2_AUDIT={audit} bash offload/launch/start_exp.sh {ck.CONFIG}')
        lines.append('  exact32000 warmup,32 generated tokens; discard timing')
        if audit:
            lines += ['  native measured q; tails36*q+{1,127,128,129}; prompt327675+5 normal length finish',
                '  matched isolated A/A then ON/OFF numeric floor at512/32000/150000/300000; needles6/6 at150K/300K',
                '  native recovery/scheduler/provenance evidence (missing = NOT RUN/HOLD); no supplied native fault adapter']
        else:
            lines.append('  cold exact'+ '/'.join(map(str,contexts)) +' x3,1500 output reserve; shared salt pairs; cache0; timing/spreads informational')
    if screen:
        lines.insert(1,'SCREEN ONLY; adoptable=false. Missing native evidence = NOT RUN; continue API gates.')
        lines = [line.replace('native measured q;', 'native measured q if supplied; otherwise ESTIMATED CPU q=4092, tails unverified;')
                 .replace('missing = NOT RUN/HOLD','missing = NOT RUN, continue SCREEN') for line in lines]
    return '\n'.join(lines)


def validate_qualification(directory: Path, expected: str, *, screen: bool = False) -> None:
    qualification = ck.read(directory/'qualification.json')
    ck.require((qualification.get('diagnostics') == 'PASS' or
        (screen and qualification.get('diagnostics') == 'PASS-SCREEN' and
         qualification.get('screen') is True and qualification.get('adoptable') is False)) and
        qualification.get('manifest_sha256') == ck.digest(ROOT/'offload/upstream_adopt/MANIFEST.json') and
        qualification.get('config_sha256') == ck.digest(ROOT/ck.CONFIG) and
        qualification.get('worker_source') == ck.WORKER and
        qualification.get('experiment_commit') == expected,
        'prior diagnostics absent/drifted')
    artifacts = record(qualification['artifact_sha256'])
    ck.require(bool(artifacts),'prior qualification evidence absent')
    for name,sha in artifacts.items():
        path = directory/name
        ck.require(path.resolve().is_relative_to(directory.resolve()) and
                   ck.digest(path) == sha,'prior qualification evidence changed')
    for arm in ('diag-off','diag-on'):
        identity = ck.read(directory/f'{arm}-identity.json')
        ids = [str(r['id']) for r in ck.rows(identity['ranks'])]
        for stage,complete in ((('q',False),('complete',True)) if screen else (('complete',True),)):
            receipt = directory/f'{arm}-native-{stage}.json'
            if receipt.is_file():
                ck.native(ck.read(receipt),arm,ids,str(qualification['manifest_sha256']),
                          Path(str(qualification['native_evidence_root'])),complete=complete)
            elif screen and qualification.get('diagnostics') == 'PASS-SCREEN':
                missing = directory/f'{arm}-native-NOT-RUN.json'
                ck.require(missing.name in artifacts,'screen missing native record not bound')
                data = ck.read(missing)
                ck.require(data.get('status') == 'NOT RUN' and data.get('screen') is True and
                           data.get('adoptable') is False and data.get('checks') ==
                           {key:'NOT RUN' for key in ck.NATIVE_CHECKS},'invalid screen NOT RUN record')
            else:
                ck.require(False,'prior native receipt missing')


def coordinator() -> None:
    values = [os.environ[k] for k in ('D2_COORDINATOR','WINDOW_COORDINATOR') if k in os.environ]
    ck.require(bool(values) and all(v == 'Deepseek/coordinator' for v in values), 'coordinator identity required')


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    mode = ap.add_mutually_exclusive_group()
    mode.add_argument('--execute',action='store_true')
    mode.add_argument('--dry-run',action='store_true')
    ap.add_argument('--screen',action='store_true',help='explicit non-adoptable screen; missing native evidence is NOT RUN')
    ap.add_argument('--phase',choices=('all','diagnostics','measured'),default='all')
    ap.add_argument('--contexts',type=int,nargs='+',choices=(150000,300000),default=None,
                    help='measured-only context subset; default both')
    ap.add_argument('--out',type=Path)
    ap.add_argument('--remote-repo',default='dsv41-deploy-offload')
    ap.add_argument('--expected-experiment-commit')
    ap.add_argument('--native-evidence',type=Path)
    ap.add_argument('--native-wait-seconds',type=float,default=0,help='bounded coordinator evidence checkpoint; default fail immediately')
    ap.add_argument('--qualification',type=Path,help='completed prior diagnostic run, for measured-only window')
    ap.add_argument('--data-dir',type=Path,default=Path('/opt/dsv41/data/bench-data'))
    args = argparse.Namespace(out=None,execute=False,screen=False)
    supervisor: Supervisor | None = None
    created_out: Path | None = None
    try:
        args = ap.parse_args()
        ck.require(args.contexts is None or args.phase == 'measured', '--contexts requires --phase measured')
        contexts = tuple(args.contexts) if args.contexts is not None else (150000, 300000)
        ck.measured_contexts(contexts)
        if not args.execute:
            print(plan(args.phase,screen=args.screen,contexts=contexts))
            return 0
        coordinator()
        ck.require(0 <= args.native_wait_seconds <= 900,'native wait must be0..900 seconds')
        ck.require(args.out is not None,'fresh --out required')
        out = cast(Path,args.out)
        ck.require(out.resolve().is_relative_to(CAMPAIGN) and out.resolve() != CAMPAIGN,'output must be private campaign directory')
        ck.require(re.fullmatch('[A-Za-z0-9_./-]+',args.remote_repo) is not None and
                   not args.remote_repo.startswith('/') and '..' not in args.remote_repo,'unsafe remote checkout')
        ck.require(re.fullmatch('[0-9a-f]{40}',args.expected_experiment_commit or '') is not None,'full reviewed experiment commit required')
        local_revision = subprocess.check_output(['git','rev-parse','HEAD'],cwd=ROOT,text=True).strip()
        ck.require(local_revision == args.expected_experiment_commit,'local reviewed checkout revision differs')
        ck.require(not subprocess.check_output(['git','status','--porcelain','--untracked-files=no'],cwd=ROOT,text=True).strip(),
                   'local tracked source is dirty; review and commit before execution')
        ck.require(args.screen or args.native_evidence is not None or args.phase == 'measured','native evidence directory required; unavailable = HOLD')
        if args.phase != 'measured' and not args.screen:
            ck.require(args.native_evidence.is_dir(), 'native apparatus/evidence directory unavailable; NOT RUN/HOLD')
        if args.phase == 'measured':
            ck.require(args.qualification is not None,'prior diagnostic qualification required')
            validate_qualification(args.qualification,args.expected_experiment_commit,screen=args.screen)
        # Validate corpus availability on CPU before any start. Never download/substitute filler.
        from offload.prefill_trace.window_requests import corpus
        corpus(args.data_dir)
        out.mkdir(parents=False,exist_ok=False)
        created_out = out
        # Read the already-open HOLD once; its timestamp, not runner start time,
        # owns the reservation. Reinvocation never renews an existing window.
        hold = subprocess.run([*SSH,'sparkhead','cat ~/dsv41-deploy/HOLD'],
            check=True,text=True,capture_output=True,timeout=8).stdout
        opened,submit,hard = ck.window_deadlines(hold,time.time())
        save(out/'deadlines.json',{'opened':opened,'submission':submit,'hard':hard},screen=args.screen)
        (out/'coordinator-next.sh').write_text('#!/usr/bin/env bash\n# Review saved rollback.json first; production still coordinator-owned.\n'+RESTORE+'\n')
        def checkpoint(data: dict[str, Json]) -> None:
            with (out/'checkpoints.jsonl').open('a') as handle:
                handle.write(json.dumps(ck.screen_record(data,args.screen),allow_nan=False)+'\n')
        def interrupted(signum: int, _frame: object) -> None:
            raise RuntimeError(f'interrupted by signal{signum}')
        for sig in (signal.SIGTERM,signal.SIGINT,signal.SIGHUP): signal.signal(sig,interrupted)
        supervisor = Supervisor(out,hard)
        supervisor.begin()
        live = Live(out,args.remote_repo,args.expected_experiment_commit,hard,submit,
                    args.native_evidence or out,args.data_dir,supervisor,args.native_wait_seconds,screen=args.screen)
        live.preflight()
        result = sequence(live,args.phase,checkpoint,screen=args.screen,contexts=contexts)
        result.update({'manifest_sha256':ck.digest(ROOT/'offload/upstream_adopt/MANIFEST.json'),
            'config_sha256':ck.digest(ROOT/ck.CONFIG),'worker_source':ck.WORKER,
            'experiment_commit':args.expected_experiment_commit,
            'native_evidence_root':str(live.native_root.resolve()),
            'artifact_sha256':{p.name:ck.digest(p) for p in out.iterdir() if p.is_file() and
                not p.name.startswith('memory-') and p.suffix in ('.json','.jsonl')}})
        save(out/'summary.json',result,screen=args.screen)
        if result.get('diagnostics') in ('PASS','PASS-SCREEN'): save(out/'qualification.json',result,screen=args.screen)
        print('Sequence complete. HOLD; inspect private evidence and restore production.')
        return 0
    except (Exception,KeyboardInterrupt) as exc:
        if created_out is not None and not (created_out/'STOP').exists():
            with (created_out/'STOP').open('x') as handle:
                handle.write(f'{type(exc).__name__}: {exc}\n')
        print(str(exc),file=sys.stderr)
        return 1
    finally:
        try:
            if supervisor is not None: supervisor.close()
            if created_out is not None and (created_out/'requests.jsonl').is_file():
                events = [record(cast(Json,json.loads(x))) for x in (created_out/'requests.jsonl').read_text().splitlines()]
                save(created_out/'accounting.json',ck.accounting(events),screen=args.screen)
        finally:
            print('Coordinator restoration commands (runner never executes these):\n'+RESTORE)
            print('Require K1 START VERIFIED; cap327680; KV2502600; greedy6/6 vs K1-production/greedy-post.jsonl; PRODUCTION HOLD OK before close.')


if __name__ == '__main__':
    raise SystemExit(main())