"""Deadline-bounded validation-only selection across independent SSH campaigns.

Launch: python scripts/fleet_campaign.py --plan /absolute/fleet-plan.json
Inspect/stop: python scripts/fleet_campaign.py --campaign /absolute/run --status [--stop]
A stopped coordinator can resume with --campaign RUN --resume; selection is frozen
once selection.json exists. Candidate campaigns have their own training deadlines.
"""
import argparse
from collections import defaultdict
from datetime import datetime, timezone
import fcntl
import hashlib
import json
import math
import os
from pathlib import Path
import re
import shlex
import shutil
import signal
import statistics
import subprocess
import sys
import time
import urllib.request

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))
from scripts import launch_24h as runner
from selection import SELECTION_METRIC, validation_selection

RUN_ROOT = Path.home() / 'ai/opensysone/runs'
SSH = ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=8', '-o',
       'ServerAliveInterval=5', '-o', 'ServerAliveCountMax=2', '-o', 'StrictHostKeyChecking=yes']
SMALL_FILES = ('manifest.json', 'data_filter.json', 'summary.json',
               'correctness_initial.json', 'correctness_final.json')
PROTECTED_SPLITS = ('validation', 'calibration', 'test', 'holdout')


def utc():
    return datetime.now(timezone.utc).isoformat()


def checksum(path):
    digest = hashlib.sha256()
    with Path(path).open('rb') as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b''):
            digest.update(chunk)
    return digest.hexdigest()


def timestamp(value):
    parsed = datetime.fromisoformat(value.replace('Z', '+00:00'))
    if parsed.tzinfo is None:
        raise ValueError('Deadlines must include an explicit timezone')
    return parsed.timestamp()


def validate_plan(plan, now=None):
    now = time.time() if now is None else now
    cutoff, deadline = timestamp(plan['selection_cutoff']), timestamp(plan['final_deadline'])
    if deadline - cutoff < 7200:
        raise ValueError('Reserve at least two hours after selection for finalization')
    if deadline <= now or deadline > now + 24 * 3600 + 10:
        raise ValueError('Final deadline must be future and at most 24 hours away')
    if not 1 <= plan.get('port', 18081) <= 65535:
        raise ValueError('Invalid API port')
    if not 1 <= plan.get('inference_max_tokens', 1024) <= 32768:
        raise ValueError('Invalid inference token limit')
    for field in ('dataset', 'reference_predictions'):
        if not Path(plan[field]).is_absolute():
            raise ValueError(f'{field} must be absolute')
    if plan.get('reference_sha256') and checksum(plan['reference_predictions']) != plan['reference_sha256']:
        raise ValueError('Frozen reference-validation checksum mismatch')
    names = set()
    for candidate in plan['candidates']:
        name, host = candidate['name'], candidate.get('host', 'local')
        if not re.fullmatch(r'[a-zA-Z0-9_-]+', name) or name in names:
            raise ValueError('Candidate names must be unique safe path components')
        names.add(name)
        if not re.fullmatch(r'[a-zA-Z0-9_.@-]+', host) or host.startswith('-'):
            raise ValueError('Invalid SSH host')
        for field in ('project', 'python', 'campaign', 'training'):
            if not Path(candidate[field]).is_absolute():
                raise ValueError(f'Candidate {field} must be absolute')
        if 'dataset' in candidate:
            verify_candidate_dataset(candidate, plan, candidate['dataset'])
        for directory in candidate.get('evidence_dirs', []):
            if not Path(directory).is_absolute():
                raise ValueError('Evidence directory must be absolute')
    if not names or not any(c.get('host', 'local') == 'local' for c in plan['candidates']):
        raise ValueError('Include a local GX10 candidate as the reachable fallback')
    return cutoff, deadline


def verify_candidate_dataset(candidate, plan, actual_dataset):
    """Only explicitly registered train-only dataset changes are comparable."""
    expected = Path(candidate.get('dataset', plan['dataset']))
    if not expected.is_absolute():
        raise ValueError('Candidate dataset override must be absolute')
    if Path(actual_dataset) != expected:
        raise ValueError('Candidate uses an unregistered or incorrect dataset location')
    if 'dataset' not in candidate:
        return None  # Historical candidates retain their exact original behavior.
    baseline = Path(plan['dataset'])
    manifests = {path: json.loads((path/'manifest.json').read_text()) for path in (baseline, expected)}
    protected = {}
    for split in PROTECTED_SPLITS:
        hashes = []
        for path in (baseline, expected):
            recorded = manifests[path].get('split_sha256', {}).get(split)
            if not isinstance(recorded, str) or not re.fullmatch(r'[0-9a-f]{64}', recorded):
                raise ValueError('Dataset lacks protected split checksum: ' + split)
            source = path/f'{split}.jsonl'
            if not source.is_file() or checksum(source) != recorded:
                raise ValueError('Protected split source hash mismatch: ' + split)
            hashes.append(recorded)
        if hashes[0] != hashes[1]:
            raise ValueError('Dataset override changes protected split: ' + split)
        protected[split] = hashes[0]
    return {'dataset': str(expected), 'reference_dataset': str(baseline),
            'protected_split_sha256': protected,
            'scope': 'Only training data may differ; protected source bytes verified without reading labels or predictions'}


def verify_selected_dataset(selected, plan):
    candidates = [candidate for candidate in plan['candidates'] if candidate['name'] == selected['name']]
    if len(candidates) != 1:
        raise ValueError('Selected candidate is not uniquely registered in the fleet plan')
    return verify_candidate_dataset(candidates[0], plan, selected['config']['dataset'])


def identity(rows):
    result = {}
    for row in rows:
        key = row['id']
        if key in result:
            raise ValueError('Duplicate validation ID')
        result[key] = (row['group'], row['family'], row['target'], row['choices'])
    return result


def validation_score(rows, reference):
    if len(rows) != 512 or identity(rows) != identity(reference):
        raise ValueError('Selection requires the identical 512 validation decisions')
    grouped = defaultdict(list)
    correct = 0
    for row in rows:
        logits, probabilities, logs = row['logits'], row['probabilities'], row['log_probabilities']
        target = row['target']
        if not isinstance(target, int) or isinstance(target, bool):
            raise ValueError('Invalid target index')
        if len(logits) < 2 or not 0 <= target < len(logits):
            raise ValueError('Invalid candidate count/target')
        if len(probabilities) != len(logits) or len(logs) != len(logits) or len(row['choices']) != len(logits):
            raise ValueError('Inconsistent prediction vector lengths')
        if any(not isinstance(v, (int, float)) or not math.isfinite(v)
               for values in (logits, probabilities, logs) for v in values):
            raise ValueError('Non-finite predictions')
        maximum = max(logits)
        logsum = math.log(sum(math.exp(v - maximum) for v in logits))
        expected_logs = [v - maximum - logsum for v in logits]
        expected_probabilities = [math.exp(v) for v in expected_logs]
        if any(p < 0 or p > 1 for p in probabilities) or abs(sum(probabilities) - 1) > 2e-6:
            raise ValueError('Invalid probabilities')
        if any(abs(a-b) > 2e-5 for a,b in zip(logs, expected_logs)) or any(
                abs(a-b) > 2e-6 for a,b in zip(probabilities, expected_probabilities)):
            raise ValueError('Saved probabilities/log probabilities disagree with logits')
        grouped[row['family']].append(-expected_logs[target])
        correct += max(range(len(logits)), key=logits.__getitem__) == target
    if len(grouped) != 4 or any(len(values) != 128 for values in grouped.values()):
        raise ValueError('Expected four validation families with 128 decisions each')
    return {'macro_nll': statistics.mean(statistics.mean(v) for v in grouped.values()),
            'accuracy': correct / len(rows), 'count': len(rows),
            'per_family_nll': {k: statistics.mean(v) for k,v in grouped.items()}}


def verify_candidate(saved, predictions, reference):
    if saved.get('format') != 'opensysone-adapter-v1' or not saved.get('trainable_state'):
        raise ValueError('Missing durable trained artifact')
    if 'temperature' in saved:
        raise ValueError('Selection requires the raw validation-trained artifact')
    score = validation_score(predictions, reference)
    stored = saved['best_validation_macro_nll']
    if not math.isfinite(stored) or abs(score['macro_nll'] - stored) > 2e-5:
        raise ValueError('Validation score does not match the durable best checkpoint')
    if saved['config'].get('validation_per_family') != 128:
        raise ValueError('Checkpoint was not selected on all 512 validation decisions')
    if saved.get('selection_metric') != SELECTION_METRIC:
        raise ValueError('Checkpoint selection policy is unsupported; legacy raw-NLL artifacts require migration')
    selected = validation_selection(predictions)
    stored_selection = saved.get('best_validation_selection_score')
    if not isinstance(stored_selection,(int,float)) or not math.isfinite(stored_selection) or abs(
            selected['score'] - stored_selection) > 1e-8:
        raise ValueError('Crossfit selection score does not match the durable best checkpoint')
    score.update(selection_metric=SELECTION_METRIC,selection_score=selected['score'],selection=selected)
    return score


def choose_candidate(eligible):
    complete = [row for row in eligible if row.get('eligible') is True]
    if not complete:
        raise RuntimeError('No verified durable candidate is eligible')
    return min(complete, key=lambda row: (row['metrics']['selection_score'], row['name']))


def verify_correctness(directory):
    expected = {'branch_chunks_1_probability_max_abs', 'branch_chunks_2_probability_max_abs',
                'branch_chunks_4_probability_max_abs', 'question_isolation_probability_max_abs',
                'candidate_permutation_probability_max_abs', 'repeat_probability_max_abs'}
    final = directory/'correctness_final.json'
    path = final if final.exists() else directory/'correctness_initial.json'
    if not path.exists():
        raise ValueError('Candidate lacks its source-run FP32 correctness evidence')
    checks = json.loads(path.read_text())
    tolerance = checks.get('tolerance_probability_abs')
    if not expected.issubset(checks) or not isinstance(tolerance,(int,float)) or not 0 < tolerance <= 1e-4:
        raise ValueError('Incomplete or weakened FP32 correctness gate')
    if any(not isinstance(value,(int,float)) or not math.isfinite(value) or not 0 <= value <= tolerance
           for key,value in checks.items() if key.endswith('max_abs')):
        raise ValueError('Candidate failed FP32 correctness gate')
    return path


def verify_compatibility(saved):
    from training_model import ADAPTER_VERSION, PROMPT_VERSION
    if saved.get('adapter_version') != ADAPTER_VERSION or saved.get('prompt_version') != PROMPT_VERSION:
        raise ValueError('Artifact prompt/adapter implementation is incompatible')
    dataset = Path(saved['config']['dataset'])
    manifest = json.loads((dataset/'manifest.json').read_text())
    # Hash opaque source files for provenance; never parse reserved split labels
    # or predictions when selecting candidates.
    for split,expected in manifest['split_sha256'].items():
        if checksum(dataset/f'{split}.jsonl') != expected:
            raise ValueError('Frozen dataset source hash mismatch: ' + split)
    signature = hashlib.sha256(json.dumps({'data':manifest['split_sha256'],
        'model':saved['model_provenance'],'implementation':checksum(ROOT/'training_model.py'),
        'max_tokens':saved['config']['max_tokens']},sort_keys=True).encode()).hexdigest()
    if signature != saved['data_signature']:
        raise ValueError('Artifact dataset/model/implementation signature is incompatible')


def command_on(candidate, arguments, timeout=30):
    command = arguments if candidate.get('host', 'local') == 'local' else [
        *SSH, candidate['host'], shlex.join(arguments)]
    result = subprocess.run(command, capture_output=True, text=True, timeout=timeout)
    if result.returncode:
        raise RuntimeError(f'{candidate["name"]}: command exited {result.returncode}: {result.stderr[-1500:]}')
    return result.stdout


def candidate_status(candidate, stop=False):
    arguments = [candidate['python'], str(Path(candidate['project'])/'scripts/campaign_status.py'),
                 '--campaign', candidate['campaign']]
    if stop:
        arguments.append('--stop')
    output = command_on(candidate, arguments)
    # The stop helper emits one explanatory line before its JSON object.
    return json.loads(output[output.index('{'):])


def stop_candidate(candidate):
    report = candidate_status(candidate)
    end = time.monotonic() + 100
    while any(report.get(k) for k in ('supervisor_running', 'child_running', 'api_running')):
        if time.monotonic() >= end:
            raise RuntimeError('Recorded campaign did not stop within 100 seconds')
        try:
            candidate_status(candidate, stop=True)
        except Exception:
            # The independent training deadline can end the supervisor between
            # inspection and TERM. Accept only a fresh, verified stopped state.
            report = candidate_status(candidate)
            if not any(report.get(k) for k in ('supervisor_running', 'child_running', 'api_running')):
                return report
            raise
        time.sleep(5)
        report = candidate_status(candidate)
    return report


def copy_evidence(candidate, source, destination, include_best=False):
    destination.mkdir(parents=True, exist_ok=False)
    patterns = [*SMALL_FILES, 'best_validation_predictions.json', 'initial_validation_predictions.json', 'resumed_initial_predictions.json',
                'validation_step_*_predictions.json']
    if include_best:
        patterns.append('best.pt')
    if candidate.get('host', 'local') == 'local':
        for pattern in patterns:
            for path in Path(source).glob(pattern):
                if path.is_file():
                    shutil.copy2(path, destination/path.name)
    else:
        command = ['rsync', '-a', '--protect-args', '--timeout=60', '-e', shlex.join(SSH)]
        command.extend('--include=' + pattern for pattern in patterns)
        command.extend(['--exclude=*', candidate['host'] + ':' + source.rstrip('/') + '/', str(destination) + '/'])
        subprocess.run(command, check=True, capture_output=True, text=True, timeout=180)


def snapshot_candidate(candidate, destination, reference):
    import torch
    destination.mkdir(parents=True, exist_ok=False)
    remote_best = str(Path(candidate['training'])/'best.pt')
    before = command_on(candidate, ['sha256sum', remote_best]).split()[0]
    copy_evidence(candidate, candidate['training'], destination/'training', include_best=True)
    correctness_path = verify_correctness(destination/'training')
    for index,directory in enumerate(candidate.get('evidence_dirs', [])):
        copy_evidence(candidate, directory, destination/f'evidence_{index}')
    checkpoint = destination/'training/best.pt'
    after = command_on(candidate, ['sha256sum', remote_best]).split()[0]
    if before != after or checksum(checkpoint) != before:
        raise ValueError('Candidate checkpoint changed during snapshot or transfer')
    saved = torch.load(checkpoint, map_location='cpu', weights_only=False)
    verify_compatibility(saved)
    step = saved['step']
    expected_name = f'validation_step_{step:06d}_predictions.json' if step else 'initial_validation_predictions.json'
    files = list(destination.glob('*/best_validation_predictions.json')) + list(destination.glob('*/' + expected_name))
    # Never use a resumed initial prediction without proof of its original step.
    # Inherited best artifacts require their parent's named validation evidence.
    verified = []
    errors = []
    for path in files:
        try:
            predictions = json.loads(path.read_text())
            metrics = verify_candidate(saved, predictions, reference)
            verified.append((path, metrics))
        except (KeyError, TypeError, ValueError) as error:
            errors.append(str(error))
    if not verified:
        raise ValueError('Durable best has no matching complete validation evidence: ' + '; '.join(errors))
    evidence, metrics = verified[0]
    model_path = Path(saved['config']['model'])
    if not model_path.is_dir():
        raise ValueError('Pinned base model is not available at the saved GX10 path')
    if json.loads((model_path/'opensysone-provenance.json').read_text()) != saved['model_provenance']:
        raise ValueError('Available GX10 base model pin does not match the artifact')
    record = {'name': candidate['name'], 'host': candidate.get('host', 'local'), 'eligible': True,
              'source_campaign': candidate['campaign'], 'source_training': candidate['training'],
              'checkpoint': str(checkpoint), 'checkpoint_sha256': before,
              'step': step, 'training_source_commit': saved['source_commit'],
              'model_provenance': saved['model_provenance'], 'data_signature': saved['data_signature'],
              'config': saved['config'], 'prediction_path': str(evidence),
              'prediction_sha256': checksum(evidence), 'metrics': metrics,
              'correctness_path': str(correctness_path),
              'source_evidence_dirs': candidate.get('evidence_dirs', []),
              'evidence_sha256': {str(p.relative_to(destination)):checksum(p)
                                  for p in destination.glob('*/*.json')},
              'selection_scope': 'Fixed four-fold temperature-crossfit validation macro-family NLL; no reserved calibration/test/holdout predictions read'}
    runner.write_json(destination/'verified.json', record)
    return record


def recorded_process(pid, expected):
    if not pid or not expected:
        return False
    try:
        return Path(f'/proc/{pid}/cmdline').read_bytes().split(b'\0')[:-1] == [v.encode() for v in expected]
    except FileNotFoundError:
        return False


def stage(campaign, state, name, arguments, deadline):
    if runner.STOP or time.time() >= deadline:
        raise RuntimeError('Stop/deadline reached before ' + name)
    code = runner.child_stage(campaign, state, name, arguments, deadline)
    if runner.STOP:
        raise InterruptedError('Coordinator stop requested')
    if code != 0:
        raise RuntimeError(f'{name} exited {code}; durable artifacts retained')


def deploy(campaign, state, plan, selected, deadline):
    evaluation = campaign/'evaluation'
    model = evaluation/'model.pt'
    metrics_path = evaluation/'metrics.json'
    # A resume may reuse only a completed evaluation of the same frozen selection.
    complete = metrics_path.exists() and json.loads(metrics_path.read_text()).get('status') == 'complete'
    if complete:
        manifest = json.loads((evaluation/'manifest.json').read_text())
        if manifest['checkpoint_sha256'] != selected['checkpoint_sha256']:
            raise ValueError('Completed evaluation belongs to a different selection')
        if manifest.get('data_signature') != selected['data_signature']:
            raise ValueError('Completed evaluation data signature differs from selection')
        for name in ('experiment.py','training_model.py'):
            if manifest.get('source_sha256',{}).get(name) != checksum(ROOT/name):
                raise ValueError('Core evaluation source changed since completed evaluation')
    else:
        if evaluation.exists():
            evaluation.rename(campaign/('evaluation_interrupted_' + datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ')))
        stage(campaign,state,'evaluation',[sys.executable,str(ROOT/'experiment.py'),'finalize',
              '--checkpoint',selected['checkpoint'],'--dataset',selected['config']['dataset'],'--output',str(evaluation)],deadline-120)
    result = json.loads(metrics_path.read_text())
    if result.get('status') != 'complete' or checksum(model) != result.get('model_sha256'):
        raise ValueError('Evaluation/model completion or checksum mismatch')
    token_args = ['--max-tokens', str(plan.get('inference_max_tokens', 1024))]
    stage(campaign,state,'harness_check',[sys.executable,str(ROOT/'jev_harness.py'),'--backend','local',
          '--checkpoint',str(model),'--request',str(ROOT/'examples/jev_request.json'),*token_args],deadline-60)
    api_command = [sys.executable,str(ROOT/'jev_harness.py'),'--backend','serve','--checkpoint',str(model),
                   '--port',str(plan.get('port',18081)),*token_args]
    with (campaign/'api.log').open('a') as log:
        api = subprocess.Popen(api_command,cwd=ROOT,stdin=subprocess.DEVNULL,stdout=log,
                               stderr=subprocess.STDOUT,start_new_session=True)
    state.update(stage='deploying',api_pid=api.pid,api_command=api_command,
                 url=f'http://127.0.0.1:{plan.get("port",18081)}',api_ready=False)
    runner.write_json(campaign/'state.json',state)
    try:
        ready_deadline = min(deadline, time.time()+60)
        while time.time() < ready_deadline and not runner.STOP and api.poll() is None:
            state['heartbeat_utc'] = utc()
            runner.write_json(campaign/'state.json',state)
            try:
                headers = {}
                if os.environ.get('OPENSYSONE_API_KEY'):
                    headers['Authorization'] = 'Bearer ' + os.environ['OPENSYSONE_API_KEY']
                with urllib.request.urlopen(urllib.request.Request(state['url']+'/health',headers=headers),timeout=3) as response:
                    health = json.load(response)
                if health.get('status') == 'ready' and health.get('checkpoint') == str(model):
                    state['api_ready'] = True
                    break
            except (OSError,ValueError):
                pass
            time.sleep(2)
        if not state['api_ready']:
            raise RuntimeError('Selected-model API failed to become ready before deadline')
        if runner.STOP or time.time() >= deadline:
            raise InterruptedError('Stop/deadline reached before final API verification')
        from jev_harness import validate_response
        payload = json.loads((ROOT/'examples/jev_request.json').read_text())
        headers['Content-Type'] = 'application/json'
        request = urllib.request.Request(state['url']+'/v1/systemone',
                                         data=json.dumps(payload).encode(),headers=headers,method='POST')
        with urllib.request.urlopen(request,timeout=min(30,max(0.1,deadline-time.time()))) as response:
            api_result = json.load(response)
        validate_response(payload,api_result)
        runner.write_json(campaign/'api_probe.json',{'request':payload,'response':api_result,
                                                   'verified_utc':utc(),'checkpoint_sha256':checksum(model)})
        if runner.STOP or time.time() >= deadline:
            raise InterruptedError('Stop/deadline reached before publishing deployment')
        pointer = {'model':str(model),'sha256':checksum(model),'campaign':str(campaign),
                   'evaluation':str(metrics_path),'selection':str(campaign/'selection.json'),
                   'inference_max_tokens':plan.get('inference_max_tokens',1024),
                   'model_id':health.get('model'),'finalized_utc':utc()}
        directory = Path.home()/'ai/opensysone/deploy'
        directory.mkdir(parents=True,exist_ok=True)
        runner.write_json(directory/'current.json',pointer)
        runner.write_json(campaign/'deployment.json',pointer)
        state.update(status='complete',stage='serving',model=str(model),finished_utc=utc())
        runner.write_json(campaign/'state.json',state)
    except BaseException:
        if api.poll() is None:
            api.terminate()
            try:
                api.wait(timeout=5)
            except subprocess.TimeoutExpired:
                api.kill()
                api.wait(timeout=5)
        raise


def run_locked(campaign):
    campaign = Path(campaign).resolve()
    plan = json.loads((campaign/'plan.json').read_text())
    cutoff, deadline = validate_plan(plan)
    runner.STOP = False
    runner.CHILD = None
    signal.signal(signal.SIGTERM,runner.stop)
    signal.signal(signal.SIGINT,runner.stop)
    Path('/proc/self/oom_score_adj').write_text('0')
    state = {'campaign':str(campaign),'supervisor_pid':os.getpid(),
             'supervisor_command':[sys.executable,str(Path(__file__).resolve()),'--campaign',str(campaign),'--run'],
             'status':'running','stage':'waiting_for_selection','started_utc':utc(),
             'source_commit':subprocess.check_output(['git','rev-parse','HEAD'],cwd=ROOT,text=True).strip(),
             'source_sha256':checksum(Path(__file__))}
    runner.write_json(campaign/'state.json',state)
    try:
        selection_path = campaign/'selection.json'
        if not selection_path.exists():
            while time.time() < cutoff and not runner.STOP:
                state['heartbeat_utc'] = utc()
                runner.write_json(campaign/'state.json',state)
                time.sleep(min(5,max(0.01,cutoff-time.time())))
            if runner.STOP:
                raise InterruptedError('Coordinator stopped; independently bounded candidate jobs remain recorded in plan')
            state.update(stage='stopping_candidates',candidates={})
            runner.write_json(campaign/'state.json',state)
            # Stop all reachable jobs before snapshotting. Unreachable remote jobs
            # are excluded; an unverified local job must block GPU finalization.
            stopped = []
            for candidate in plan['candidates']:
                if runner.STOP:
                    raise InterruptedError('Coordinator stopped during candidate shutdown')
                try:
                    report = stop_candidate(candidate)
                    state['candidates'][candidate['name']] = {'stop_status':report}
                    stopped.append(candidate)
                except Exception as error:
                    state['candidates'][candidate['name']] = {'eligible':False,'error':str(error)}
                    if candidate.get('host','local') == 'local':
                        raise RuntimeError('Cannot establish that GX10 candidate stopped') from error
                runner.write_json(campaign/'state.json',state)
            if plan.get('reference_sha256') and checksum(plan['reference_predictions']) != plan['reference_sha256']:
                raise ValueError('Reference validation changed while awaiting selection')
            reference = json.loads(Path(plan['reference_predictions']).read_text())
            rows = [{'name':name,**record} for name,record in state['candidates'].items()
                    if record.get('eligible') is False]
            attempt = campaign/('selection_attempt_' + datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ'))
            attempt.mkdir()
            for candidate in stopped:
                if runner.STOP:
                    raise InterruptedError('Coordinator stopped during candidate snapshots')
                try:
                    row = snapshot_candidate(candidate,attempt/candidate['name'],reference)
                    dataset_proof = verify_candidate_dataset(candidate, plan, row['config']['dataset'])
                    if dataset_proof is not None:
                        row['dataset_compatibility'] = dataset_proof
                except Exception as error:
                    row = {'name':candidate['name'],'eligible':False,'error':str(error)}
                rows.append(row)
                runner.write_json(attempt/'candidates.json',rows)
                state['heartbeat_utc'] = utc()
                runner.write_json(campaign/'state.json',state)
            selected = choose_candidate(rows)
            if runner.STOP:
                raise InterruptedError('Coordinator stopped before freezing selection')
            runner.write_json(selection_path,{'selected':selected,'candidates':rows,'selected_utc':utc(),
                                             'reference_sha256':checksum(plan['reference_predictions'])})
        else:
            selected = json.loads(selection_path.read_text())['selected']
        if checksum(selected['checkpoint']) != selected['checkpoint_sha256']:
            raise ValueError('Frozen selection checkpoint changed')
        # Also recheck the explicit registration/protected bytes when resuming a
        # frozen selection. Full artifact/data signatures were checked at snapshot.
        verify_selected_dataset(selected, plan)
        if runner.STOP:
            raise InterruptedError('Coordinator stop requested')
        state.update(stage='acquiring_gx10_lock',selected=selected['name'])
        runner.write_json(campaign/'state.json',state)
        with (RUN_ROOT/'.smoke.lock').open('a') as lock:
            fcntl.flock(lock,fcntl.LOCK_EX | fcntl.LOCK_NB)
            deploy(campaign,state,plan,selected,deadline)
        return 0
    except Exception as error:
        state.update(status='interrupted' if runner.STOP else 'failed',error=str(error),finished_utc=utc())
        runner.write_json(campaign/'state.json',state)
        print(json.dumps(state),flush=True)
        return 1


def run(campaign):
    RUN_ROOT.mkdir(parents=True,exist_ok=True)
    with (RUN_ROOT/'.fleet.lock').open('a') as lock:
        try:
            fcntl.flock(lock,fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            print('Another fleet coordinator holds .fleet.lock',file=sys.stderr,flush=True)
            return 1
        try:
            return run_locked(campaign)
        except Exception as error:
            state = {'campaign':str(Path(campaign).resolve()),'supervisor_pid':os.getpid(),
                     'status':'failed','stage':'startup','error':str(error),'finished_utc':utc()}
            runner.write_json(Path(campaign)/'state.json',state)
            print(json.dumps(state),flush=True)
            return 1


def inspect(campaign, stop=False):
    state = json.loads((campaign/'state.json').read_text())
    flags = {key:recorded_process(state.get(key+'_pid'),state.get(key+'_command'))
             for key in ('supervisor','child','api')}
    if stop:
        for key in ('supervisor','child','api'):
            if flags[key]:
                os.kill(state[key+'_pid'],signal.SIGTERM)
                break
        else:
            raise RuntimeError('No exact command-verified fleet process remains')
    print(json.dumps({'state':state,'running':flags},indent=2))
    return flags


def detach(campaign):
    environment = os.environ.copy()
    environment.update(PYTHONUNBUFFERED='1',TOKENIZERS_PARALLELISM='false',HF_HUB_DISABLE_XET='1',
                       PYTORCH_CUDA_ALLOC_CONF='expandable_segments:True')
    with (campaign/'supervisor.log').open('a') as log:
        process = subprocess.Popen([sys.executable,str(Path(__file__).resolve()),'--campaign',str(campaign),'--run'],
            cwd=ROOT,stdin=subprocess.DEVNULL,stdout=log,stderr=subprocess.STDOUT,env=environment,start_new_session=True)
    (RUN_ROOT/'LAST_FLEET_CAMPAIGN').write_text(str(campaign)+'\n')
    print(json.dumps({'campaign':str(campaign),'supervisor_pid':process.pid}))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--plan')
    parser.add_argument('--campaign')
    parser.add_argument('--run',action='store_true',help=argparse.SUPPRESS)
    parser.add_argument('--status',action='store_true')
    parser.add_argument('--stop',action='store_true')
    parser.add_argument('--resume',action='store_true')
    args = parser.parse_args()
    if args.plan:
        if args.campaign or args.run or args.status or args.stop or args.resume:
            parser.error('--plan cannot be combined with campaign control')
        plan = json.loads(Path(args.plan).read_text())
        plan.setdefault('reference_sha256',checksum(plan['reference_predictions']))
        validate_plan(plan)
        campaign = RUN_ROOT/(datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ')+'-fleet')
        campaign.mkdir(parents=True,exist_ok=False)
        runner.write_json(campaign/'plan.json',plan)
        detach(campaign)
    else:
        if not args.campaign:
            parser.error('--plan or --campaign is required')
        campaign = Path(args.campaign).resolve()
        if args.run:
            code = run(campaign)
            (campaign/'exit_code').write_text(str(code)+'\n')
            raise SystemExit(code)
        if args.resume:
            validate_plan(json.loads((campaign/'plan.json').read_text()))
            flags = inspect(campaign)
            if any(flags.values()):
                raise RuntimeError('Stop existing recorded fleet processes before resume')
            detach(campaign)
        else:
            inspect(campaign,stop=args.stop)


if __name__ == '__main__':
    main()