#!/usr/bin/env python3
"""Independent RCM workflow: exclusive attempts; no cleanup or production drivers."""
import argparse
import ctypes
import ctypes.util
import datetime as dt
import hashlib
import itertools
import json
import os
from pathlib import Path
import re
import shlex
import shutil
import subprocess
import sys
import traceback
import uuid

# Keep a single core module when invoked as a script.
if __name__ == '__main__':
    sys.modules['rcms'] = sys.modules[__name__]

HERE = Path(__file__).resolve().parent
UTC = dt.timezone.utc


def preserve_deletions():
    """Kernel-enforced no-op for unlink/rmdir, inherited across fork/exec.

    Return success WITHOUT executing deletion so legacy Fortran cleanup does
    not abort a valid run. No missing-library or unsupported-kernel fallback.
    """
    libname = ctypes.util.find_library('seccomp')
    need(libname, 'libseccomp is required for deletion protection')
    lib = ctypes.CDLL(libname, use_errno=True)
    lib.seccomp_init.argtypes = [ctypes.c_uint32]
    lib.seccomp_init.restype = ctypes.c_void_p
    lib.seccomp_syscall_resolve_name.argtypes = [ctypes.c_char_p]
    lib.seccomp_syscall_resolve_name.restype = ctypes.c_int
    lib.seccomp_rule_add.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_int, ctypes.c_uint]
    lib.seccomp_rule_add.restype = ctypes.c_int
    lib.seccomp_load.argtypes = [ctypes.c_void_p]
    lib.seccomp_load.restype = ctypes.c_int
    lib.seccomp_release.argtypes = [ctypes.c_void_p]
    ctx = lib.seccomp_init(0x7fff0000)  # SCMP_ACT_ALLOW
    need(ctx, 'Cannot allocate deletion filter')
    try:
        for syscall in [b'unlink', b'unlinkat', b'rmdir']:
            nr = lib.seccomp_syscall_resolve_name(syscall)
            need(nr >= 0, 'Unknown deletion syscall: ' + str(syscall))
            # SCMP_ACT_ERRNO(0): syscall is skipped, success returned.
            need(lib.seccomp_rule_add(ctx, 0x00050000, nr, 0) == 0,
                 'Cannot create deletion filter for ' + str(syscall))
        need(lib.seccomp_load(ctx) == 0, 'Kernel refused deletion filter; execution blocked')
    finally:
        lib.seccomp_release(ctx)


def guarded(executable):
    return [sys.executable, str(HERE / 'rcms.py'), '_exec_guard', str(executable)]


def stamp():
    return dt.datetime.now(UTC).strftime('%Y%m%dT%H%M%SZ') + '-' + uuid.uuid4().hex[:8]


def put(path, value):
    path = Path(path)
    with path.open('x') as f:
        f.write(value if isinstance(value, str) else json.dumps(value, indent=2))


def copy_new(src, dst):
    with Path(src).open('rb') as inf, Path(dst).open('xb') as out:
        shutil.copyfileobj(inf, out)


def need(ok, message):
    if not ok:
        raise RuntimeError(message)


def date10(s):
    need(bool(re.fullmatch(r'\d{10}', s)), 'Expected YYYYMMDDHH')
    return dt.datetime.strptime(s, '%Y%m%d%H')


def iso(s):
    return date10(s).strftime('%Y-%m-%d_%H:00:00')


def times(start, end):
    t, stop = date10(start), date10(end)
    while t <= stop:
        yield t.strftime('%Y%m%d%H')
        t += dt.timedelta(hours=6)


def grib_complete(path):
    """Check every GRIB2 message boundary, catching truncated downloads."""
    size = Path(path).stat().st_size
    with Path(path).open('rb') as f:
        pos = 0
        while pos < size:
            f.seek(pos)
            head = f.read(16)
            need(len(head) == 16 and head[:4] == b'GRIB' and head[7] == 2,
                 'Invalid GRIB2 header: ' + str(path))
            length = int.from_bytes(head[8:16], 'big')
            need(length >= 20 and pos + length <= size, 'Truncated GRIB2: ' + str(path))
            f.seek(pos + length - 4)
            need(f.read(4) == b'7777', 'GRIB2 end marker missing: ' + str(path))
            pos += length
    need(size > 0, 'Empty GRIB2: ' + str(path))


def inventory(cfg, cycle, end=None):
    need(cycle[-2:] == '00', 'This preserved operational configuration supports cycle 00 only')
    date10(cycle)
    folder = Path(cfg['input_root']).resolve() / cycle[:8]
    entries = {}
    for p in sorted(folder.glob('*.grb2')):
        m = re.fullmatch(r'(flxf|pgbf)(\d{10})\.01\.(\d{10})\.grb2', p.name)
        if not m:
            continue
        need(m[3] == cycle, 'Mixed initialization in ' + str(p))
        need(p.stat().st_size > 0, 'Empty GRIB: ' + str(p))
        entries[(m[1], m[2])] = {'path': str(p.resolve()), 'size': p.stat().st_size,
                                         'mtime_ns': p.stat().st_mtime_ns}
    need(entries, 'No GRIB inputs in ' + str(folder))
    last = end or max(t for _, t in entries)
    need(date10(last) > date10(cycle), 'End must follow initialization')
    expected = list(times(cycle, last))
    need(expected[-1] == last, 'End is not on six-hour grid')
    selected = []
    for t in expected:
        for family in ('flxf', 'pgbf'):
            need((family, t) in entries, 'Missing ' + family + ' at ' + t)
            grib_complete(entries[(family, t)]['path'])
            selected.append(dict(entries[(family, t)], family=family, time=t))
    return last, selected


def render(text, start, end):
    for prefix, val in [('S', start), ('E', end)]:
        for key, part in [('YYYY', val[:4]), ('MM', val[4:6]), ('DD', val[6:8]), ('HH', val[8:10])]:
            text = text.replace(prefix + '_' + key, part)
    for key, val in [('D_RESTART', '.false.'), ('S_RESTART', start), ('E_RESTART', end),
                     ('S_DATE', start), ('E_DATE', end)]:
        text = text.replace(key, val)
    return text


def set_nml(text, key, value):
    pattern = r'(?mi)^(\s*' + re.escape(key) + r'\s*=)[^\n]*'
    new, n = re.subn(pattern, lambda m: m[1] + ' ' + value + ',', text)
    need(n == 1, 'Namelist key must occur once: ' + key)
    return new


class Runner:
    def __init__(self, work, utilities=None):
        self.utilities = utilities or {}
        self.work = Path(work)
        self.logs = self.work / 'commands'
        self.logs.mkdir()
        self.number = 0

    def run(self, args, cwd=None):
        self.number += 1
        args = [str(a) for a in args]
        args[0] = self.utilities.get(args[0], args[0])
        stem = self.logs / ('%05d' % self.number)
        put(str(stem) + '.json', {'argv': args, 'cwd': str(cwd or self.work)})
        with Path(str(stem) + '.log').open('x') as out:
            result = subprocess.run(guarded(args[0]) + args[1:], cwd=str(cwd or self.work), stdout=out,
                                    stderr=subprocess.STDOUT)
        need(result.returncode == 0, 'Command failed (%s): %s; log: %s.log' %
             (result.returncode, ' '.join(shlex.quote(str(arg)) for arg in args), stem))
        return Path(str(stem) + '.log').read_text(errors='replace')

    def nc(self, path):
        need(Path(path).is_file() and Path(path).stat().st_size > 0, 'Missing/empty NetCDF: ' + str(path))
        return self.run(['ncdump', '-h', path])

    def axis(self, path, start, end, hours):
        self.nc(path)
        output = self.run(['cdo', '-s', 'showtimestamp', path])
        actual = re.findall(r'\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d', output)
        expected = []
        t, stop = date10(start), date10(end)
        while t <= stop:
            expected.append(t.strftime('%Y-%m-%dT%H:%M:%S'))
            t += dt.timedelta(hours=hours)
        need(actual == expected, 'Time axis incomplete or different: ' + str(path))


def template(name):
    return (HERE / 'templates' / name).read_text()


def inputs_unchanged(man):
    for entry in man['inputs']:
        s = Path(entry['path']).stat()
        need(s.st_size == entry['size'] and s.st_mtime_ns == entry['mtime_ns'],
             'Input changed after submission: ' + entry['path'])


def link(src, dst):
    src = Path(src).resolve(strict=True)
    Path(dst).symlink_to(src)


def preflight(cfg, model):
    commands = ['ncdump', 'cdo', 'ncpdq', 'ncatted', 'ncks', 'ncwa', 'nccopy', 'ncrename']
    commands += ['mpirun'] if model == 'wrf' else ['wgrib2', 'srun']
    for command in commands:
        binary = cfg.get('utilities', {}).get(command, command)
        need(shutil.which(binary), 'Missing command after environment setup: ' + binary)
    exes = ([Path(cfg['wps_dir']) / (n + '.exe') for n in ['ungrib', 'metgrid']] +
            [Path(cfg['wrf_dir']) / 'run' / (n + '.exe') for n in ['real', 'wrf']] +
            [Path(cfg['arwpost'])]) if model == 'wrf' else [
                Path(cfg['regcm_bin']) / n for n in ['sstCLM45', 'icbcCLM45', 'regcmMPICLM45', 'sigma2pCLM45']]
    for p in exes:
        need(p.is_file() and os.access(str(p), os.X_OK), 'Missing executable: ' + str(p))


def wrf_integrate(man, work, r):
    cfg, start, end = man['config'], man['cycle'], man['end']
    if cfg.get('operational_layout'):
        for entry in man['inputs']:
            link(entry['path'], work / 'input' / Path(entry['path']).name)
    fields = {}
    for family, prefix in [('flxf', 'FLXF'), ('pgbf', 'PGBF')]:
        stage = work / ('ungrib_' + prefix)
        stage.mkdir()
        nml = render(template('wps.nml'), start, end).replace('FFFF', prefix).replace('TOPO_DIR', cfg['topo_dir'])
        put(stage / 'namelist.wps', nml)
        link(Path(cfg['wps_dir']) / 'ungrib/Variable_Tables/Vtable.CFSR', stage / 'Vtable')
        codes = itertools.product('ABCDEFGHIJKLMNOPQRSTUVWXYZ', repeat=3)
        for entry in (e for e in man['inputs'] if e['family'] == family):
            link(entry['path'], stage / ('GRIBFILE.' + ''.join(next(codes))))
        r.run([Path(cfg['wps_dir']) / 'ungrib.exe'], stage)
        log = (stage / 'ungrib.log').read_text(errors='replace')
        need('Successful completion' in log, 'ungrib did not report success')
        for t in times(start, end):
            p = stage / (prefix + ':' + iso(t)[:13])
            need(p.is_file() and p.stat().st_size > 0, 'Missing intermediate: ' + str(p))
        fields[prefix] = stage
    met = work / 'metgrid'
    met.mkdir()
    nml = render(template('wps.nml'), start, end).replace('FFFF', 'FLXF').replace('TOPO_DIR', cfg['topo_dir'])
    nml = set_nml(nml, 'fg_name', "'../ungrib_FLXF/FLXF', '../ungrib_PGBF/PGBF'")
    put(met / 'namelist.wps', nml)
    link(cfg['geo_em'], met / 'geo_em.d01.nc')
    link(Path(cfg['wps_dir']) / 'metgrid/METGRID.TBL.ARW', met / 'METGRID.TBL')
    r.run([Path(cfg['wps_dir']) / 'metgrid.exe'], met)
    need('Successful completion' in (met / 'metgrid.log').read_text(), 'metgrid success marker missing')
    return wrf_from_met(man, work, r, met)


def wrf_from_met(man, work, r, met):
    cfg, start, end = man['config'], man['cycle'], man['end']
    real = work / 'real'
    real.mkdir()
    raw = real
    import recovery
    levels = recovery.check_met(man, met, r)
    for t in times(start, end):
        p = met / ('met_em.d01.' + iso(t) + '.nc')
        link(p, raw / p.name)
    # Copies of static tables prevent any executable from modifying shared tables.
    runtime_assets = set()
    for pattern in ['*.TBL', 'ozone*', 'RRTMG*']:
        runtime_assets.update((Path(cfg['wrf_dir']) / 'run').glob(pattern))
    for p in sorted(runtime_assets):
        copy_new(p, raw / p.name)
    copy_new(Path(cfg['wrf_dir']) / 'run/CAMtr_volume_mixing_ratio.RCP4.5', raw / 'CAMtr_volume_mixing_ratio')
    put(raw / 'namelist.input', render(template('wrf.nml'), start, end).replace('NUM_MET_LEVEL', levels))
    put(work / 'PREPROCESS_SUCCESS.json', {'end': end})
    r.run(['mpirun', '-np', '10'] + guarded(Path(cfg['wrf_dir']) / 'run/real.exe'), raw)
    need('SUCCESS COMPLETE REAL' in (raw / 'rsl.error.0000').read_text(errors='replace'), 'REAL success marker missing')
    import recovery
    raw = recovery.output_dir(man, 'wrf', work)
    runtime = work / 'wrf'
    runtime.mkdir()
    (runtime / 'output').symlink_to(raw)
    for p in real.iterdir():
        if p.name.startswith(('rsl.', 'met_em.', 'wrfrst', 'wrfout', 'wrfxtrm')) or p.suffix == '.log':
            continue
        if p.is_file():
            if p.name != 'namelist.input':
                copy_new(p, runtime / p.name)
    nml = (real / 'namelist.input').read_text()
    nml = nml.replace('&time_control', "&time_control\n history_outname = 'output/wrfout_d<domain>_<date>',\n rst_outname = 'output/wrfrst_d<domain>_<date>',", 1)
    nml = set_nml(nml, 'auxhist3_outname', "'output/wrfxtrm_d<domain>_<date>'")
    put(runtime / 'namelist.input', nml)
    for name in ['wrfinput_d01', 'wrfbdy_d01', 'wrflowinp_d01']:
        r.nc(runtime / name)
    r.run(['mpirun', '-np', str(man['tasks']['wrf'])] + guarded(Path(cfg['wrf_dir']) / 'run/wrf.exe'), runtime)
    need('SUCCESS COMPLETE WRF' in (runtime / 'rsl.error.0000').read_text(errors='replace'), 'WRF success marker missing')
    for t in times(start, end):
        r.nc(raw / ('wrfout_d01_' + iso(t)))
    need(iso(end) in r.run(['ncdump', '-v', 'Times', raw / ('wrfout_d01_' + iso(end))]), 'WRF last timestamp mismatch')
    return {'raw': str(raw), 'start': start, 'end': end, 'marker': 'SUCCESS COMPLETE WRF'}


def regcm_integrate(man, work, r):
    cfg, start, end = man['config'], man['cycle'], man['end']
    dataset = work / 'Dataset'
    dataset.mkdir()
    for name in ['CLM45', 'SST', 'SURFACE']:
        link(Path(cfg['regcm_dataset']) / name, dataset / name)
    if cfg.get('operational_layout'):
        (Path(cfg['regcm_dataset']) / 'CFS' / start).mkdir()  # Reserve exclusively.
        link(Path(cfg['regcm_dataset']) / 'CFS', dataset / 'CFS')
    data = dataset / 'CFS' / start / '01'
    (data / 'SST').mkdir(parents=True)
    (data / 'PLEV').mkdir()
    specs = [('flxf', ':TMP:surface', 'SST/sst'), ('pgbf', ':TMP:', 'PLEV/air'),
             ('pgbf', ':RH:', 'PLEV/rhum'), ('pgbf', ':UGRD:', 'PLEV/uwnd'),
             ('pgbf', ':VGRD:', 'PLEV/vwnd'), ('pgbf', ':HGT:', 'PLEV/hgt')]
    for family, match, name in specs:
        dest = data / (name + '.' + start + '.nc')
        need(not dest.exists(), 'Refusing existing converted file')
        for e in (x for x in man['inputs'] if x['family'] == family):
            cmd = ['wgrib2', e['path'], '-match', match]
            if family == 'pgbf':
                cmd += ['-match', 'mb:']
            cmd += ['-nc_table', HERE / 'templates/prc.table']
            if dest.exists():
                cmd += ['-append']  # Only the new file owned by this attempt.
            r.run(cmd + ['-nc4', '-netcdf', dest])
        r.axis(dest, start, end, 6)
    import recovery
    inp, raw = work / 'input', recovery.output_dir(man, 'regcm', work)
    if not cfg.get('operational_layout'):
        inp.mkdir()
    for p in sorted(Path(cfg['regcm_static']).glob('d01_*')):
        if p.is_file():
            copy_new(p, inp / p.name)
    nml = render(template('regcm.nml'), start, end)
    for key, value in [('dirter', 'input'), ('dirglob', 'input'), ('inpter', 'Dataset'), ('inpglob', 'Dataset'), ('dirout', str(raw))]:
        nml = set_nml(nml, key, "'%s'" % value)
    put(work / 'namelist.in', nml)
    r.run([Path(cfg['regcm_bin']) / 'sstCLM45', 'namelist.in'])
    icbc = r.run([Path(cfg['regcm_bin']) / 'icbcCLM45', 'namelist.in'])
    need('Successfully completed ICBC' in icbc, 'ICBC success marker missing')
    r.axis(inp / 'd01_SST.nc', start, end, 6)
    monthkeys = sorted(set(t[:6] for t in times(start, end)))
    for month in monthkeys:
        key = start if month == start[:6] else month + '0100'
        r.nc(inp / ('d01_ICBC.' + key + '.nc'))
    put(work / 'PREPROCESS_SUCCESS.json', {'end': end})
    return regcm_from_input(man, work, r, raw)


def regcm_from_input(man, work, r, raw):
    cfg, start, end = man['config'], man['cycle'], man['end']
    log = r.run(['srun', '-n', str(man['tasks']['regcm'])] + guarded(Path(cfg['regcm_bin']) / 'regcmMPICLM45') + ['namelist.in'])
    need('RegCM V4 simulation successfully reached end' in log, 'RegCM success marker missing')
    for family in ['ATM', 'SRF', 'STS', 'RAD']:
        files = sorted(raw.glob('d01_' + family + '.*.nc'))
        need(files, 'No RegCM ' + family + ' output')
        for p in files:
            r.nc(p)
        # Daily STS may label the last completed day, other streams must reach end.
        if family != 'STS':
            axis = r.run(['cdo', '-s', 'showtimestamp', files[-1]])
            need(iso(end).replace('_', 'T') in axis, 'RegCM ' + family + ' did not reach end')
    return {'raw': str(raw), 'start': start, 'end': end,
            'marker': 'RegCM V4 simulation successfully reached end'}


def pack(r, src, var, output, work, drop=(), divide=None, grid=False, levels=False):
    """Distinct intermediates; no overwrite flags or cleanup."""
    stage = work / var
    stage.mkdir()
    cmd = ['cdo', '-s']
    if grid:
        cmd += ['setgridtype,lonlat']
    if levels:
        cmd += ['-sellevel,1000.,925.,850.,700.,600.,500.,300.,200.,100.']
    if divide:
        cmd += ['-divc,' + str(divide)]
    cmd += ['-selname,' + var, src, stage / 'selected.nc']
    r.run(cmd)
    current = stage / 'selected.nc'
    header = r.nc(current)
    dims = [d for d in drop if re.search(r'\b' + re.escape(d) + r'\s*=\s*\d+', header)]
    if dims:
        r.run(['ncwa', '-a', ','.join(dims), current, stage / 'averaged.nc'])
        current = stage / 'averaged.nc'
        coordinate_vars = [d for d in dims if re.search(r'\b(?:float|double|int|short)\s+' + re.escape(d) + r'\s*\(', header)]
        if 'bnds' in dims and 'time_bnds' in header:
            coordinate_vars.append('time_bnds')
        if coordinate_vars:
            r.run(['ncks', '-C', '-x', '-v', ','.join(coordinate_vars), current, stage / 'without_dimensions.nc'])
            current = stage / 'without_dimensions.nc'
    if divide:
        r.run(['ncatted', '-a', 'long_name,' + var + ',o,c,Geopotential height (m)', current, stage / 'height.nc'])
        current = stage / 'height.nc'
    r.run(['ncpdq', current, stage / 'packed.nc'])
    r.run(['ncatted', '-a', '_FillValue,,o,f,-32767', stage / 'packed.nc', stage / 'attributes.nc'])
    r.run(['cdo', '-s', '-f', 'nc4c', '-z', 'zip_9', 'copy', stage / 'attributes.nc', output])
    r.nc(output)


def wrf_post(man, raw, work, r):
    cfg, start, end = man['config'], man['cycle'], man['end']
    arw = work / 'arwpost'
    arw.mkdir()
    # ARWpost stores filenames in CHARACTER(LEN=132): use short relative links.
    (arw / 'input').mkdir()
    for p in sorted(raw.glob('wrfout_d01_*')):
        link(p, arw / 'input' / p.name)
    nml = render(template('arwpost.nml'), start, end).replace('IN_FFFF', 'input/wrfout_d01_*').replace('OUT_FFFF', 'wrfout')
    put(arw / 'namelist.ARWpost', nml)
    r.run([cfg['arwpost']], arw)
    r.run(['cdo', '-s', '-f', 'nc', 'import_binary', arw / 'wrfout.ctl', arw / 'wrfout.nc'])
    r.axis(arw / 'wrfout.nc', start, end, 6)
    products, interim = work / 'products', work / 'intermediate'
    products.mkdir()
    interim.mkdir()
    names = r.run(['cdo', '-s', 'showname', arw / 'wrfout.nc']).split()
    required = {'slp','u10m','v10m','u','v','w','rh2','rh','tsk','t2','tk','theta','geopt','q2','qvapor','rainc','rainnc','olr'}
    need(required.issubset(names), 'ARWpost variables missing: ' + str(required - set(names)))
    for var in names:
        need(re.fullmatch(r'[A-Za-z][A-Za-z0-9_]*', var), 'Unexpected variable name')
        pack(r, arw / 'wrfout.nc', var, products / (var + '.nc'), interim,
             drop=('lev_2',), divide=9.8 if var == 'geopt' else None)
    # Preserve all original wrfxtrm files; manipulate separate copies only.
    xt = work / 'extremes'
    xt.mkdir()
    prepared = []
    for p in sorted(raw.glob('wrfxtrm_d01_*')):
        stage = xt / ('step%04d' % len(prepared))
        stage.mkdir()
        r.run(['nccopy', '-k', '2', p, stage / 'copy.nc'])
        r.run(['ncrename', '-d', 'west_east,lon', '-d', 'south_north,lat', stage / 'copy.nc', stage / 'renamed.nc'])
        copy_new(stage / 'renamed.nc', stage / 'coordinates.nc')
        r.run(['ncks', '-A', '-v', 'lon,lat', cfg['domain'], stage / 'coordinates.nc'])
        selected = stage / 'selected.nc'
        r.run(['cdo', '-s', 'selname,T2MAX,T2MIN,T2MEAN,U10MAX,V10MAX,RAINCVMEAN,RAINNCVMEAN', stage / 'coordinates.nc', selected])
        datepart = p.name[len('wrfxtrm_d01_'):]
        r.run(['cdo', '-s', '-setcalendar,standard', '-settaxis,' + datepart[:10] + ',' + datepart[11:] + ',6hour',
               '-setreftime,1979-01-01,00:00,hours', selected, stage / 'timed.nc'])
        prepared.append(stage / 'timed.nc')
    need(prepared, 'No WRF extremes outputs')
    r.run(['cdo', '-s', '-r', 'mergetime'] + prepared + [xt / 'merged.nc'])
    r.run(['ncpdq', xt / 'merged.nc', xt / 'packed.nc'])
    r.run(['ncatted', '-a', '_FillValue,,o,f,-32767', xt / 'packed.nc', xt / 'attributes.nc'])
    r.run(['cdo', '-s', '-f', 'nc4c', '-z', 'zip_9', 'copy', xt / 'attributes.nc', products / 'xtrm.nc'])
    for p in sorted(products.glob('*.nc')):
        # Extremes may start at first output interval instead of initial state.
        if p.name != 'xtrm.nc':
            r.axis(p, start, end, 6)
        else:
            r.nc(p)
            axis = r.run(['cdo', '-s', 'showtimestamp', p])
            need(iso(end).replace('_', 'T') in axis, 'Extremes did not reach end')
    return products


def regcm_post(man, raw, work, r):
    products = work / 'products'
    products.mkdir()
    for d in ['daily', '3hourly', '6hourly']:
        (products / d).mkdir()
    specs = [('ATM', '6hourly', 'ua va wa ta hus rh hgt mslp ts'),
             ('SRF', '3hourly', 'pr prc uas vas ts huss hurs psl evspsbl zmla hfss hfls'),
             ('STS', 'daily', 'pr prmax tas tasmax tasmin'), ('RAD', '6hourly', 'rlut prw')]
    for family, interval, variables in specs:
        stage = work / family
        stage.mkdir()
        sources = sorted(raw.glob('d01_' + family + '.*.nc'))
        need(sources, 'Missing source family ' + family)
        if family == 'ATM':
            converted = []
            for p in sources:
                sub = stage / p.stem
                sub.mkdir()
                r.run([Path(man['config']['regcm_bin']) / 'sigma2pCLM45', p], sub)
                out = sub / (p.stem + '_pressure.nc')
                r.nc(out)
                converted.append(out)
            sources = converted
        r.run(['cdo', '-s', '-r', 'mergetime'] + sources + [stage / 'merged.nc'])
        for var in variables.split():
            target = products / interval / (var + '.nc')
            pack(r, stage / 'merged.nc', var, target, stage, drop=('bnds', 'm2', 'm10'),
                 grid=True, levels=family == 'ATM' and var not in ('mslp', 'ts'))
            out_axis = re.findall(r'\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d', r.run(['cdo', '-s', 'showtimestamp', target]))
            src_axis = re.findall(r'\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d', r.run(['cdo', '-s', 'showtimestamp', stage / 'merged.nc']))
            need(out_axis and out_axis == src_axis, 'Postprocessing changed time coverage: ' + str(target))
            step = {'daily':24,'3hourly':3,'6hourly':6}[interval]
            parsed = [dt.datetime.strptime(x, '%Y-%m-%dT%H:%M:%S') for x in out_axis]
            need(all(b-a == dt.timedelta(hours=step) for a,b in zip(parsed, parsed[1:])), 'Gaps in ' + str(target))
            need(parsed[0] <= date10(man['cycle']) + dt.timedelta(hours=step) and
                 parsed[-1] >= date10(man['end']) - dt.timedelta(hours=step), 'Truncated coverage: ' + str(target))
    return products


def worker(manifest, model, phase, token):
    manifest = Path(manifest).resolve()
    man = json.loads(manifest.read_text())
    root = manifest.parent
    need(os.environ.get('SLURM_JOB_ID'), 'Workers must run inside their submitted Slurm allocation')
    work = root / model / ('integration' if phase == 'integrate' else 'post-' + token)
    import operations as ops
    ops.require_user(man['config'])
    if phase == 'integrate' and man['config'].get('operational_layout') and model not in man.get('recovery', {}):
        target = Path(man['config']['legacy_dirs'][model]) / 'wdir' / man['cycle']
        target.mkdir()  # Refuse any pre-existing operational cycle.
        work.mkdir()
        (target / 'input').mkdir()
        (work / 'input').symlink_to(target / 'input')
    else:
        work.mkdir()  # Existing attempt is never reused, including after failure.
    os.environ['TMPDIR'] = str(work / 'tmp')
    Path(os.environ['TMPDIR']).mkdir()
    put(work / 'STARTED.json', {'time': stamp(), 'job_id': os.environ['SLURM_JOB_ID']})
    try:
        preserve_deletions()
        preflight(man['config'], model)
        r = Runner(work, man['config'].get('utilities'))
        if phase == 'integrate':
            inputs_unchanged(man)
            if model in man.get('recovery', {}):
                import recovery
                result = recovery.resume(man, model, work, r)
            else:
                result = (wrf_integrate if model == 'wrf' else regcm_integrate)(man, work, r)
        else:
            import recovery
            source = man.get('recovery', {}).get(model, {})
            if source.get('action') == 'ADOPT_POST':
                raw = Path(source['raw'])
                man = recovery.validate_raw(man, model, raw, r)
                put(work/'RAW_VALIDATED.json', {'raw':str(raw),'end':man['end']})
            else:
                success = root / model / 'integration/SUCCESS.json'
                need(success.is_file(), 'Integration success record missing; postprocessing refused')
                record = json.loads(success.read_text())
                raw = Path(record['raw'])
                expected = (Path(source['work'])/'output' if source.get('action') == 'RESUME' else
                            Path(man['config']['legacy_dirs'][model]) / 'wdir' / man['cycle'] / 'output'
                            if man['config'].get('operational_layout') else root / model / 'integration/raw')
                need(raw.resolve() == expected.resolve(), 'Raw path mismatch')
            products = (wrf_post if model == 'wrf' else regcm_post)(man, raw, work, r)
            result = {'products': str(products), 'files': [str(p.relative_to(products)) for p in sorted(products.rglob('*.nc'))]}
        put(work / 'SUCCESS.json', dict(result, time=stamp()))
    except BaseException as exc:
        put(work / 'FAILED.json', {'time': stamp(), 'error': str(exc), 'traceback': traceback.format_exc()})
        raise
    if phase == 'post' and man['config'].get('current_result_root'):
        try:
            ops.publish(root, model, work)
        except Exception as exc:
            put(work / 'ARCHIVE_FAILED.json', {'error': str(exc), 'time': stamp()})
            raise


def validate_config(cfg, models):
    for key in ['input_root', 'wrf_dir', 'wps_dir', 'arwpost', 'regcm_bin', 'regcm_static', 'regcm_dataset', 'geo_em', 'domain']:
        p = Path(cfg[key])
        need(p.is_absolute() and p.exists(), 'Missing configured path: ' + key + '=' + str(p))
        need(not any(c in str(p) for c in "'\"\n\r"), 'Unsupported quote/newline in path')
    for model in models + ['post']:
        s = cfg['slurm'][model]
        need(isinstance(s['nodes'], int) and s['nodes'] > 0 and isinstance(s['tasks_per_node'], int) and s['tasks_per_node'] > 0, 'Invalid Slurm resources')
        need(bool(re.fullmatch(r'\d+(?:-\d+)?(?::\d+){1,2}', s['time'])), 'Invalid walltime')
    for name in ['wrf.nml','wps.nml','arwpost.nml','regcm.nml','prc.table','wrf.env','regcm.env']:
        need((HERE / 'templates' / name).is_file(), 'Missing snapshot ' + name)


def job_script(root, man, model, phase, token):
    cfg = man['config']
    res = cfg['slurm']['post' if phase == 'post' else model]
    name = model + '-' + phase + '-' + token
    slurm_name = 'rcms-' + model + '-' + man.get('cycle', 'unknown') + '-' + phase + '-' + token
    script = root / (name + '.sh')
    env = root / 'runtime/templates' / (model + '.env')
    module_lines = '\n'.join('module add ' + shlex.quote(m) for m in cfg['post_modules'])
    # Environment snapshots configure libraries only; no production run scripts.
    body = '#!/bin/bash -l\nset -e\nsource ' + shlex.quote(str(env)) + '\n' + module_lines + '\nset -euo pipefail\n'
    body += 'cd ' + shlex.quote(str(root)) + '\n'
    body += 'exec ' + ' '.join(shlex.quote(str(arg)) for arg in [man['python'], str(root / 'runtime/rcms.py'), '_worker',
                                 str(root / 'manifest.json'), model, phase, token]) + '\n'
    put(script, body)
    cmd = ['sbatch', '--parsable', '--job-name=' + slurm_name, '--nodes=' + str(res['nodes']),
           '--ntasks-per-node=' + str(res['tasks_per_node']), '--time=' + res['time'],
           '--chdir=' + str(root), '--output=' + str(root / (name + '-%j.out')),
           '--error=' + str(root / (name + '-%j.err'))]
    if cfg.get('partition'):
        cmd += ['--partition=' + cfg['partition']]
    return script, cmd


def submit_job(root, man, model, phase, token, dependency=None, dependency_type='afterok'):
    need(dependency_type in ('afterok', 'afterany'), 'Invalid dependency type')
    script, cmd = job_script(root, man, model, phase, token)
    if dependency:
        cmd += ['--dependency=' + dependency_type + ':' + dependency, '--kill-on-invalid-dep=yes']
    cmd.append(str(script))
    prefix = root / (model + '-' + phase + '-' + token)
    put(str(prefix) + '.submit-command.json', cmd)
    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
    put(str(prefix) + '.submission.json', {'returncode':result.returncode, 'stdout':result.stdout, 'stderr':result.stderr})
    need(result.returncode == 0, 'Submission failed; previous submissions remain recorded at ' + str(root))
    job = result.stdout.strip().split(';')[0]
    need(job.isdigit(), 'Unrecognized sbatch result; inspect submission record before retrying')
    put(str(prefix) + '.job.json', {'job_id': job, 'model':model, 'phase':phase, 'dependency':dependency,
                                              'slurm_name': next(x.split('=',1)[1] for x in cmd if x.startswith('--job-name='))})
    print(model, phase, 'job', job, flush=True)
    return job


def status(root):
    root = Path(root).resolve()
    man = json.loads((root / 'manifest.json').read_text())
    print('Run:', root, 'Cycle:', man['cycle'], 'End:', man['end'])
    for model in man['models']:
        for stage in sorted((root / model).iterdir()):
            state = 'SUCCESS' if (stage/'SUCCESS.json').exists() else 'FAILED' if (stage/'FAILED.json').exists() else 'INCOMPLETE (inspect Slurm log)'
            print(model, stage.name, state)
            if (stage/'FAILED.json').exists():
                print(json.loads((stage/'FAILED.json').read_text())['error'])
        jobs = [json.loads(p.read_text()) for p in root.glob(model + '-*.job.json')]
        for j in jobs:
            print('Recorded job:', j)
    if shutil.which('squeue'):
        ids = [json.loads(p.read_text())['job_id'] for p in root.glob('*.job.json')]
        if ids:
            subprocess.run(['squeue', '-j', ','.join(ids), '-o', '%.18i %.20j %.12T %.30R'])
    print('No SUCCESS.json means completion is unconfirmed, even if the job left the queue.')


def main():
    if len(sys.argv) > 1 and sys.argv[1] == '_worker':
        need(len(sys.argv) == 6, 'Expected _worker MANIFEST MODEL PHASE TOKEN')
        manifest, model, phase, token = sys.argv[2:]
        need(model in ('wrf', 'regcm') and phase in ('integrate', 'post'), 'Invalid worker stage')
        need(re.fullmatch(r'[A-Za-z0-9-]+', token), 'Invalid attempt token')
        worker(manifest, model, phase, token)
    else:
        import operations
        operations.main()


if __name__ == '__main__':
    try:
        if len(sys.argv) >= 3 and sys.argv[1] == '_exec_guard':
            preserve_deletions()
            os.execvp(sys.argv[2], sys.argv[2:])
        main()
    except Exception as exc:
        print('ERROR:', exc, file=sys.stderr)
        sys.exit(1)