#!/usr/bin/env python3
"""Central read-only resource monitor for WG Paid / MGTS core machines."""
import argparse, fcntl, json, os, shlex, subprocess, sys, tempfile
from pathlib import Path
from datetime import datetime, timezone
SCHEMA="router-resource-monitor-v1"
STATE_DIR=Path(os.environ.get("ROUTER_RESOURCE_MONITOR_STATE_DIR","/opt/router-ops/state/resource-monitor"))
CURRENT=STATE_DIR/"current.json"; EVENTS=STATE_DIR/"warnings.jsonl"; LOCK=STATE_DIR/"monitor.lock"
SSH_KEY=os.environ.get("ROUTER_RESOURCE_MONITOR_SSH_KEY","/home/ops/.ssh/router_ops_ed25519")
PVE_KEY="/root/.ssh/pve_to_openwrt_mgts_ed25519"
THRESHOLDS={
 "root_used_pct":{"warn":75,"crit":90},
 "root_inode_pct":{"warn":75,"crit":90},
 "tmp_used_pct":{"warn":50,"crit":75},
 "journal_kb":{"warn":786432,"crit":1572864},
 "bounded_log_bytes":{"warn":1153434,"crit":2097152},
}
HOSTS=("vm100","vm101","vm103","vm121","vm130")
def now_iso(): return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00","Z")
def run(cmd,timeout=20):
    try:
        p=subprocess.run(cmd,text=True,capture_output=True,timeout=timeout); return p.returncode,p.stdout,p.stderr
    except subprocess.TimeoutExpired as e: return 124,e.stdout or "",e.stderr or "timeout"
    except OSError as e: return 127,"",str(e)
def parse_kv(text):
    d={}
    for raw in text.splitlines():
        if "=" not in raw: continue
        k,v=raw.split("=",1); k=k.strip(); v=v.strip()
        if not k: continue
        try: d[k]=int(v)
        except ValueError: d[k]=v
    return d
def openwrt(ip,log_path):
    remote=("ROOT_USED_PCT=$(df -P / | awk 'END{gsub(\"%\",\"\",$5);print $5}'); "
            "TMP_USED_PCT=$(df -P /tmp | awk 'END{gsub(\"%\",\"\",$5);print $5}'); "
            f"LOG_BYTES=$(wc -c < {shlex.quote(log_path)} 2>/dev/null || echo 0); "
            "echo root_used_pct=$ROOT_USED_PCT; echo tmp_used_pct=$TMP_USED_PCT; echo bounded_log_bytes=$LOG_BYTES")
    inner=f"ssh -n -i {PVE_KEY} -o BatchMode=yes -o ConnectTimeout=8 root@{ip} {shlex.quote(remote)}"
    rc,o,e=run(["ssh","-o","BatchMode=yes","-o","ConnectTimeout=8","pve-mgts",inner],20); return rc,parse_kv(o),e.strip()
def debian_remote(target,sudo=False,key=None):
    remote=("ROOT_USED_PCT=$(df -P / | awk 'END{gsub(\"%\",\"\",$5);print $5}'); "
            "ROOT_INODE_PCT=$(df -Pi / | awk 'END{gsub(\"%\",\"\",$5);print $5}'); "
            "JOURNAL_KB=$(du -sk /var/log/journal 2>/dev/null | awk '{print $1}'); [ -n \"$JOURNAL_KB\" ] || JOURNAL_KB=0; "
            "echo root_used_pct=$ROOT_USED_PCT; echo root_inode_pct=$ROOT_INODE_PCT; echo journal_kb=$JOURNAL_KB")
    if sudo: remote="sudo -n sh -c "+shlex.quote(remote)
    cmd=["ssh","-o","BatchMode=yes","-o","ConnectTimeout=8"]
    if key: cmd += ["-i",key]
    cmd += [target,remote]
    rc,o,e=run(cmd,20); return rc,parse_kv(o),e.strip()
def local130():
    cmd=("ROOT_USED_PCT=$(df -P / | awk 'END{gsub(\"%\",\"\",$5);print $5}'); "
         "ROOT_INODE_PCT=$(df -Pi / | awk 'END{gsub(\"%\",\"\",$5);print $5}'); "
         "JOURNAL_KB=$(du -sk /var/log/journal 2>/dev/null | awk '{print $1}'); [ -n \"$JOURNAL_KB\" ] || JOURNAL_KB=0; "
         "echo root_used_pct=$ROOT_USED_PCT; echo root_inode_pct=$ROOT_INODE_PCT; echo journal_kb=$JOURNAL_KB")
    rc,o,e=run(["bash","-lc",cmd],10); return rc,parse_kv(o),e.strip()
def fixture_metrics(path):
    r={}
    for h in HOSTS:
        p=Path(path)/f"{h}.kv"; r[h]=(0,parse_kv(p.read_text()),"") if p.is_file() else (2,{},f"fixture_missing:{p}")
    return r
def collect(fixture=None):
    if fixture: return fixture_metrics(fixture)
    return {"vm100":openwrt("10.71.100.1","/var/log/router-wgpay-egress-local.log"),
            "vm101":openwrt("10.71.100.2","/var/log/router-egress-health-repair.log"),
            "vm103":debian_remote("ops@10.71.100.86",True,SSH_KEY),
            "vm121":debian_remote("vm121"),"vm130":local130()}
def rank(s): return {"PASS":0,"WARN":1,"CRIT":2}[s]
def evaluate(raw):
    hosts={}; warnings=[]; overall="PASS"
    for h in HOSTS:
        rc,m,err=raw[h]; hs="PASS"
        if rc:
            hs="WARN"; warnings.append({"host":h,"metric":"collection","severity":"WARN","value":rc,"detail":err[:160]})
        else:
            for metric,value in m.items():
                if metric not in THRESHOLDS or not isinstance(value,int): continue
                t=THRESHOLDS[metric]; sev="CRIT" if value>=t["crit"] else ("WARN" if value>=t["warn"] else None)
                if sev:
                    if rank(sev)>rank(hs): hs=sev
                    warnings.append({"host":h,"metric":metric,"severity":sev,"value":value,"warn":t["warn"],"crit":t["crit"]})
        if rank(hs)>rank(overall): overall=hs
        hosts[h]={"reachable":rc==0,"status":hs,**m}
        if err and rc==0: hosts[h]["stderr"]=err[:160]
    return overall,hosts,warnings
def atomic_json(path,obj):
    path.parent.mkdir(parents=True,exist_ok=True)
    fd,tmp=tempfile.mkstemp(prefix=path.name+".",dir=str(path.parent))
    try:
        with os.fdopen(fd,"w") as f: json.dump(obj,f,sort_keys=True,indent=2); f.write("\n"); f.flush(); os.fsync(f.fileno())
        os.chmod(tmp,0o600); os.replace(tmp,path)
    finally:
        try: os.unlink(tmp)
        except FileNotFoundError: pass
def trim_events():
    if not EVENTS.exists() or EVENTS.stat().st_size<=1048576: return
    lines=EVENTS.read_text(errors="replace").splitlines()[-200:]; tmp=EVENTS.with_name(EVENTS.name+".trim")
    tmp.write_text("\n".join(lines)+("\n" if lines else "")); os.chmod(tmp,0o600); os.replace(tmp,EVENTS)
def append_event(doc):
    if doc["overall"]=="PASS": return
    trim_events()
    with EVENTS.open("a") as f: f.write(json.dumps({"generated_at_utc":doc["generated_at_utc"],"overall":doc["overall"],"warnings":doc["warnings"]},sort_keys=True)+"\n")
    os.chmod(EVENTS,0o600); trim_events()
def run_monitor(fixture=None):
    STATE_DIR.mkdir(parents=True,exist_ok=True); os.chmod(STATE_DIR,0o700)
    with LOCK.open("a+") as lock:
        fcntl.flock(lock.fileno(),fcntl.LOCK_EX|fcntl.LOCK_NB)
        overall,hosts,warnings=evaluate(collect(fixture))
        doc={"schema":SCHEMA,"generated_at_utc":now_iso(),"overall":overall,"thresholds":THRESHOLDS,"hosts":hosts,"warning_count":len(warnings),"warnings":warnings}
        atomic_json(CURRENT,doc); append_event(doc)
        write_capacity(fixture)
        if fixture is None: deliver_capacity()
        print(f"RESULT={overall}_ROUTER_RESOURCE_MONITOR"); print(f"OVERALL={overall}"); print(f"WARNING_COUNT={len(warnings)}"); print(f"CURRENT={CURRENT}")
        return {"PASS":0,"WARN":10,"CRIT":20}[overall]
# Capacity telemetry is a separate private snapshot, not a health/uptime verdict.
# It runs only during the existing resource monitor's 15-minute invocation.
CAPACITY=STATE_DIR/'capacity.json'
PVE_CAPACITY_SCRIPT=r'''
import json, math, subprocess, time
from pathlib import Path

def q(path, *args):
    try:
        p=subprocess.run(['pvesh','get',path,*args,'--output-format','json'],capture_output=True,text=True,timeout=5)
        return json.loads(p.stdout) if p.returncode==0 else None
    except (OSError, ValueError, subprocess.TimeoutExpired):
        return None

def vals(rows,key):
    return sorted(float(r[key]) for r in rows if isinstance(r,dict) and isinstance(r.get(key),(int,float)) and math.isfinite(r[key]))

def summary(rows,kind):
    if not isinstance(rows,list): return {'status':'unavailable','points':0}
    rows=[r for r in rows if isinstance(r,dict)]
    ts=vals(rows,'time'); c=vals(rows,'cpu'); m=[]
    for r in rows:
        if isinstance(r.get('mem'),(int,float)) and isinstance(r.get('maxmem'),(int,float)) and r['maxmem']>0:
            x=100*r['mem']/r['maxmem']
            if math.isfinite(x): m.append(x)
    m.sort()
    def p95(v): return round(v[min(len(v)-1,math.ceil(.95*len(v))-1)],3) if v else None
    out={'status':'available' if ts and c else 'partial','points':len(rows),
         'cpu_points':len(c),'cpu_avg_pct':round(100*sum(c)/len(c),3) if c else None,
         'cpu_p95_pct':round(100*p95(c),3) if c else None,
         'mem_p95_pct':round(p95(m),3) if m else None,
         'span_hours':round((ts[-1]-ts[0])/3600,2) if len(ts)>1 else None,
         'latest_age_sec':max(0,int(time.time()-ts[-1])) if ts else None}
    for key in ('netin','netout','diskread','diskwrite'):
        out[key+'_p95_bytes_per_sec']=p95(vals(rows,key))
    return out

inventory=q('/cluster/resources','--type','vm')
result={'schema':'b3-pve-rrd-v1','source_status':'available' if isinstance(inventory,list) else 'unavailable',
        'vms':{},'nodes':{},'physical_nics':{}}
if isinstance(inventory,list):
    used=set()
    for vmid in (100,101,103,130):
        found=[x for x in inventory if isinstance(x,dict) and x.get('vmid')==vmid and x.get('type')=='qemu']
        name='vm'+str(vmid)
        if len(found)!=1:
            result['vms'][name]={'status':'unknown','inventory_matches':len(found)}
            continue
        x=found[0];node=x.get('node')
        if not isinstance(node,str) or not node or not node.replace('-','').replace('_','').isalnum():
            result['vms'][name]={'status':'unknown','node_proven':False}
            continue
        used.add(node)
        result['vms'][name]={'status':str(x.get('status','unknown'))[:16],'node':node,
                 'vcpus':x.get('maxcpu') if isinstance(x.get('maxcpu'),(int,float)) else None,
                 'allocated_memory_bytes':x.get('maxmem') if isinstance(x.get('maxmem'),int) else None,
                 'windows':{}}
        for window in ('day','week'):
            result['vms'][name]['windows'][window]=summary(q(f'/nodes/{node}/qemu/{vmid}/rrddata','--timeframe',window),'vm')
    for node in sorted(used):
        status=q(f'/nodes/{node}/status')
        obj={'windows':{},'host_cpu_count':None,'host_mem_used_pct':None}
        if isinstance(status,dict):
            obj['host_cpu_count']=(status.get('cpuinfo') or {}).get('cpus')
            mem=status.get('memory') or {}
            if isinstance(mem.get('used'),(int,float)) and isinstance(mem.get('total'),(int,float)) and mem['total']>0:
                obj['host_mem_used_pct']=round(100*mem['used']/mem['total'],2)
        for window in ('day','week'):
            obj['windows'][window]=summary(q(f'/nodes/{node}/rrddata','--timeframe',window),'node')
        result['nodes'][node]=obj
    # Only physical NIC link facts; no MAC/IP/hostname beyond Proxmox node IDs.
    for nic in sorted(Path('/sys/class/net').iterdir()):
        if not (nic/'device').exists(): continue
        try:
            state=(nic/'operstate').read_text().strip();speed=int((nic/'speed').read_text().strip())
            result['physical_nics'][nic.name]={'up':state=='up','speed_mbps':speed if speed>0 else None}
        except (OSError,ValueError):
            result['physical_nics'][nic.name]={'up':None,'speed_mbps':None}
print(json.dumps(result,separators=(',',':'),allow_nan=False))
'''
GUEST_CAPACITY_SCRIPT=r'''
import json, os
from pathlib import Path
m={}
for line in Path('/proc/meminfo').read_text().splitlines():
    k,_,v=line.partition(':')
    if k in ('MemTotal','MemAvailable','SwapTotal','SwapFree'):
        m[k]=int(v.strip().split()[0])
total=m.get('MemTotal',0)
res={'vcpus':os.cpu_count(),'mem_total_mib':total//1024,
     'mem_available_mib':m.get('MemAvailable',0)//1024,
     'mem_available_pct':round(100*m.get('MemAvailable',0)/total,2) if total else None,
     'swap_used_mib':(m.get('SwapTotal',0)-m.get('SwapFree',0))//1024,
     'psi_some_avg300':{}}
for name in ('cpu','memory','io'):
    p=Path('/proc/pressure')/name
    try:
        line=next((s for s in p.read_text().splitlines() if s.startswith('some ')),'')
        item=next((s for s in line.split() if s.startswith('avg300=')),'')
        val=float(item.split('=',1)[1]) if item else None
        res['psi_some_avg300'][name]=val if val is not None and 0<=val<=100 else None
    except (OSError,ValueError):res['psi_some_avg300'][name]=None
print(json.dumps(res,separators=(',',':')))
'''

def _capacity_run(cmd, script, timeout=40):
    try:
        p=subprocess.run(cmd,input=script,text=True,capture_output=True,timeout=timeout)
        if p.returncode or len(p.stdout)>65536:return {'status':'unavailable','failure_class':'command_or_size'}
        obj=json.loads(p.stdout)
        if not isinstance(obj,dict):return {'status':'unavailable','failure_class':'shape'}
        return obj
    except (OSError,ValueError,subprocess.TimeoutExpired):
        return {'status':'unavailable','failure_class':'transport_or_json'}

def collect_capacity(fixture=None):
    if fixture:
        p=Path(fixture)/'capacity.json'
        if p.is_file():
            obj=json.loads(p.read_text()); assert isinstance(obj,dict)
            return obj
        return {'schema':'router-capacity-snapshot-v1','generated_at_utc':now_iso(),
                'source_status':'unavailable','pve':{'status':'fixture_absent'},'guests':{}}
    pve=_capacity_run(['ssh','-T','-o','BatchMode=yes','-o','ConnectTimeout=7','pve-mgts','python3 -'],PVE_CAPACITY_SCRIPT,40)
    guests={}
    endpoints={'vm103':['ssh','-T','-o','BatchMode=yes','-o','ConnectTimeout=7','-i',SSH_KEY,'ops@10.71.100.86','python3 -'],
               'vm121':['ssh','-T','-o','BatchMode=yes','-o','ConnectTimeout=7','vm121','python3 -']}
    for host,cmd in endpoints.items():
        guests[host]=_capacity_run(cmd,GUEST_CAPACITY_SCRIPT,9)
    guests['vm130']=_capacity_run([sys.executable,'-c',GUEST_CAPACITY_SCRIPT],'',6)
    failures=int(pve.get('source_status')!='available')+sum('mem_available_pct' not in v for v in guests.values())
    return {'schema':'router-capacity-snapshot-v1','generated_at_utc':now_iso(),
            'source_status':'partial' if failures else 'available','failed_sources':failures,
            'pve':pve,'guests':guests,
            'interpretation':'historical_rrd_samples_not_instantaneous_link_utilization_or_uptime'}

def write_capacity(fixture=None):
    try:
        obj=collect_capacity(fixture)
        assert isinstance(obj,dict)
        atomic_json(CAPACITY,obj)
        print('CAPACITY_SNAPSHOT=WRITTEN')
        print('CAPACITY_SOURCE_STATUS='+str(obj.get('source_status','unknown')))
    except (OSError,ValueError,AssertionError,TypeError) as exc:
        # Capacity collection must never turn a healthy legacy warning monitor into failed service.
        atomic_json(CAPACITY,{'schema':'router-capacity-snapshot-v1','generated_at_utc':now_iso(),
                              'source_status':'unavailable','failure_class':type(exc).__name__})
        print('CAPACITY_SNAPSHOT=UNKNOWN')


# Restricted B3B4 delivery. Converts the private diagnostic snapshot into the
# VM121 capacity-ingest-v1 allowlist. No raw RRD JSON or host metadata crosses SSH.
CAPACITY_DELIVERY_ACK=STATE_DIR/'capacity-delivery-ack.json'
CAPACITY_DELIVERY_KEY=Path('/opt/router-ops/state/capacity-ingest-ssh/id_ed25519')
CAPACITY_DELIVERY_MAX=32768
CAPACITY_ALLOWED_STATES=('available','partial','unavailable','unknown')
CAPACITY_WINDOW_LIMITS={
 'cpu_avg_pct':100,'cpu_p95_pct':100,'mem_p95_pct':100,
 'netin_p95_bytes_per_sec':1e12,'netout_p95_bytes_per_sec':1e12,
 'diskread_p95_bytes_per_sec':1e12,'diskwrite_p95_bytes_per_sec':1e12,
 'latest_age_sec':86400*8,'points':100000,
}
CAPACITY_GUEST_LIMITS={
 'mem_total_mib':1e8,'mem_available_mib':1e8,'mem_available_pct':100,
 'swap_used_mib':1e8,'psi_cpu_avg300':100,'psi_memory_avg300':100,
 'psi_io_avg300':100,
}
def _cap_number(v,limit,integer=False):
    import math
    if type(v) not in (float,int) or not math.isfinite(v) or not 0<=v<=limit:
        return None
    if integer and type(v) is not int:return None
    return v

def _cap_state(s):
    return s if s in CAPACITY_ALLOWED_STATES else 'unknown'

def _cap_window(item):
    raw=item if isinstance(item,dict) else {}
    out={'status':_cap_state(raw.get('status'))}
    for name,limit in CAPACITY_WINDOW_LIMITS.items():
        out[name]=_cap_number(raw.get(name),limit,integer=name in ('points','latest_age_sec'))
    if out['status']=='available' and (out['points'] is None or out['cpu_avg_pct'] is None):
        out['status']='partial'
    return out

def build_capacity_delivery(snapshot):
    if not isinstance(snapshot,dict) or snapshot.get('schema')!='router-capacity-snapshot-v1':
        raise ValueError('unexpected source schema')
    stamp=snapshot.get('generated_at_utc')
    if not isinstance(stamp,str):raise ValueError('missing timestamp')
    # The receiving side additionally enforces UTC format and 7-day recency.
    pve=snapshot.get('pve') if isinstance(snapshot.get('pve'),dict) else {}
    pve_vms=pve.get('vms') if isinstance(pve.get('vms'),dict) else {}
    vms={}
    for ident in ('vm100','vm101','vm103','vm130'):
        vm=pve_vms.get(ident) if isinstance(pve_vms.get(ident),dict) else {}
        windows=vm.get('windows') if isinstance(vm.get('windows'),dict) else {}
        day=_cap_window(windows.get('day'));week=_cap_window(windows.get('week'))
        # A VM marked running in Proxmox is not proof of tunnel health.
        status='available' if vm.get('status')=='running' else 'unknown'
        if day['status']!='available' or week['status']!='available':
            status='partial' if status=='available' else 'unavailable'
        vms[ident]={'status':status,'vcpus':_cap_number(vm.get('vcpus'),128,integer=True),
                     'day':day,'week':week}
    raw_guests=snapshot.get('guests') if isinstance(snapshot.get('guests'),dict) else {}
    guests={}
    for ident in ('vm103','vm121','vm130'):
        raw=raw_guests.get(ident) if isinstance(raw_guests.get(ident),dict) else {}
        psi=raw.get('psi_some_avg300') if isinstance(raw.get('psi_some_avg300'),dict) else {}
        out={'status':'available' if 'mem_available_pct' in raw else 'unavailable'}
        for key,limit in CAPACITY_GUEST_LIMITS.items():
            if key.startswith('psi_'):
                rawkey=key[4:-7] # psi_cpu_avg300 -> cpu
                val=psi.get(rawkey)
            else:val=raw.get(key)
            out[key]=_cap_number(val,limit,integer=key in ('mem_total_mib','mem_available_mib','swap_used_mib'))
        if out['status']=='available' and out['mem_available_pct'] is None:out['status']='partial'
        guests[ident]=out
    nodes=pve.get('nodes') if isinstance(pve.get('nodes'),dict) else {}
    # Avoid guessing when host inventory changes: only one proven PVE node.
    item=next(iter(nodes.values())) if len(nodes)==1 else {}
    if not isinstance(item,dict):item={}
    windows=item.get('windows') if isinstance(item.get('windows'),dict) else {}
    day=_cap_window(windows.get('day'));week=_cap_window(windows.get('week'))
    node={'status':'available' if item and day['status']=='available' and week['status']=='available' else 'partial' if item else 'unknown',
          'cpu_count':_cap_number(item.get('host_cpu_count'),256,integer=True),
          'mem_used_pct':_cap_number(item.get('host_mem_used_pct'),100),
          'day_cpu_avg_pct':day['cpu_avg_pct'],'week_cpu_avg_pct':week['cpu_avg_pct']}
    nics=pve.get('physical_nics') if isinstance(pve.get('physical_nics'),dict) else {}
    up=[n for n in nics.values() if isinstance(n,dict) and n.get('up') is True]
    if len(up)==1:
        nic={'status':'available','link_up':True,'link_mbps':_cap_number(up[0].get('speed_mbps'),1e6,integer=True)}
    elif len(up)>1:
        nic={'status':'partial','link_up':True,'link_mbps':None}
    elif nics:
        nic={'status':'partial','link_up':False,'link_mbps':None}
    else:nic={'status':'unknown','link_up':None,'link_mbps':None}
    source_status=_cap_state(snapshot.get('source_status'))
    if source_status=='available' and (any(x['status']!='available' for x in vms.values()) or
                                       any(x['status']!='available' for x in guests.values()) or
                                       node['status']!='available' or nic['status']!='available'):
        source_status='partial'
    return {'schema':'wg-capacity-delivery-v1','source':'vm130_resource_monitor',
            'generated_at_utc':stamp,'source_status':source_status,
            'vms':vms,'guests':guests,'node':node,'nic':nic}

def _cap_bytes(doc):
    import hashlib
    raw=(json.dumps(doc,sort_keys=True,separators=(',',':'),allow_nan=False)+'\n').encode()
    if len(raw)>CAPACITY_DELIVERY_MAX:raise ValueError('delivery limit')
    return raw,hashlib.sha256(raw).hexdigest()

def validate_capacity_ack(raw,code,expected):
    if code!=0 or len(raw)>512:raise ValueError('ssh or ack length')
    try: a=json.loads(raw)
    except (TypeError,ValueError,UnicodeError):raise ValueError('bad ack json') from None
    if not isinstance(a,dict) or set(a)!=set(('status','sha256')):
        raise ValueError('bad ack shape')
    if a['sha256']!=expected or a['status'] not in ('accepted','duplicate'):
        raise ValueError('ack mismatch')
    return a['status']

def deliver_capacity_snapshot(*,runner=None,capacity_path=None,key=None,ack_path=None):
    import hashlib,stat as statmod
    p=capacity_path or CAPACITY
    k=key or CAPACITY_DELIVERY_KEY
    dest=ack_path or CAPACITY_DELIVERY_ACK
    st=p.lstat()
    if not statmod.S_ISREG(st.st_mode) or st.st_size>262144 or st.st_size==0 or st.st_mode&0o077:
        raise ValueError('unsafe snapshot')
    doc=build_capacity_delivery(json.loads(p.read_bytes()))
    payload,digest=_cap_bytes(doc)
    st=k.lstat()
    if not statmod.S_ISREG(st.st_mode) or st.st_mode&0o077:raise ValueError('unsafe identity')
    cmd=['ssh','-T','-o','User=capacity-ingest','-o','IdentitiesOnly=yes',
         '-o','IdentityAgent=none','-o','BatchMode=yes',
         '-o','StrictHostKeyChecking=yes','-o','ClearAllForwardings=yes',
         '-o','ConnectTimeout=8','-o','ConnectionAttempts=1',
         '-o','PreferredAuthentications=publickey','-o','PasswordAuthentication=no',
         '-o','KbdInteractiveAuthentication=no','-i',str(k),'vm121']
    if runner is None:
        result=subprocess.run(cmd,input=payload,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,timeout=18)
    else:result=runner(cmd,payload)
    ack=validate_capacity_ack(result.stdout,result.returncode,digest)
    atomic_json(dest,{'schema':'router-capacity-delivery-ack-v1',
                       'source_generated_at_utc':doc['generated_at_utc'],
                       'acknowledged_at_utc':now_iso(),'status':ack,'sha256':digest})
    return ack

def deliver_capacity():
    # Delivery errors never fail or change the legacy warning-monitor verdict.
    try:
        result=deliver_capacity_snapshot()
        print('CAPACITY_DELIVERY='+result.upper())
    except (OSError,ValueError,TypeError,UnicodeError,subprocess.TimeoutExpired):
        print('CAPACITY_DELIVERY=DEFERRED')

def main():
    ap=argparse.ArgumentParser(); ap.add_argument("--run",action="store_true"); ap.add_argument("--status",action="store_true"); ap.add_argument("--fixture"); a=ap.parse_args()
    if a.status:
        if not CURRENT.is_file(): print("RESULT=EMPTY_ROUTER_RESOURCE_MONITOR"); return 0
        sys.stdout.write(CURRENT.read_text()); return 0
    if not a.run: ap.error("use --run or --status")
    try: return run_monitor(a.fixture)
    except BlockingIOError: print("RESULT=NOOP_ROUTER_RESOURCE_MONITOR_LOCKED"); return 0
if __name__=="__main__": raise SystemExit(main())
