"""Read-only postprocessing of the six existing, authorized simulation outputs."""
import csv
import hashlib
import json
import math
from pathlib import Path
import subprocess
import datetime
import numpy as np

Q = Path('/home/frank14f/CelerisLab-release-evidence-20261020/physical-750e8a4/followup-queue')
R = Path('/home/frank14f/CelerisLab-release-prep-20261020')
SHA = '750e8a483ec47dcf472e528edc1cf2936f0e7e83'
def read(p):
    return json.loads(p.read_text())
def sha(p):
    return hashlib.sha256(p.read_bytes()).hexdigest()
def cmd(*a):
    return subprocess.check_output(a, text=True).strip()
def span(a):
    return float(np.ptp(a))
def finite(x):
    if isinstance(x, dict): return all(finite(v) for v in x.values())
    if isinstance(x, list): return all(finite(v) for v in x)
    return not isinstance(x, (int,float)) or math.isfinite(x)
def stats(rows, key):
    t=np.array([v['tstar'] for v in rows]); v=np.array([r[key] for r in rows])
    d=np.diff(v); nz=np.sign(d[d!=0])
    return dict(samples=len(v),first=float(v[0]),last=float(v[-1]),min=float(min(v)),max=float(max(v)),mean=float(np.mean(v)),range=span(v),ols_slope_per_tstar=float(np.polyfit(t-t[0],v,1)[0]),increment_sign_changes=int(np.count_nonzero(nz[1:]!=nz[:-1])),total_variation=float(np.abs(d).sum()))

state=read(Q/'state.json'); manifest=read(Q/'manifest.json'); execution=read(Q/'execution.json')
out=dict(schema=1,utc=datetime.datetime.now(datetime.timezone.utc).isoformat(),sha=SHA,evidence_root=str(Q),postprocessor_sha256=sha(Path(__file__)),scope='Offline read-only digest; no new simulation, gate edit, acceptance promotion, or performance claim',verification={},cases={})
v=out['verification']; v['all_six_completed']=len(state['jobs'])==6 and all(x['status']=='COMPLETED' for x in state['jobs'].values())
v['all_exit_zero_no_timeout']=all(x['exit_code']==0 and not x['timeout'] for x in state['jobs'].values())
v['head']=cmd('git','-C',str(R),'rev-parse','HEAD');v['git_status_porcelain']=cmd('git','-C',str(R),'status','--porcelain');v['clean_fixed_head']=v['head']==SHA and not v['git_status_porcelain']
v['tree']=cmd('git','-C',str(R),'rev-parse','HEAD^{tree}');v['src_tree']=cmd('git','-C',str(R),'rev-parse','HEAD:src')
v['tracked_source_map_matches_manifest']=all(sha(R/p)==h for p,h in manifest['files_sha256'].items());v['tracked_source_map_count']=len(manifest['files_sha256'])
v['execution_final_source_matches_manifest']=all(execution['source_after'][k]==manifest[k] for k in ['sha','tree','src_tree','git_status','git_diff_sha256','files_sha256'])
pids=[state['scheduler_pid']]+[x['pid'] for x in state['jobs'].values()]
v['tracked_processes']={}
for pid in pids:
    p=subprocess.run(['ps','-p',str(pid),'-o','pid=,ppid=,stat=,comm='],capture_output=True,text=True)
    v['tracked_processes'][str(pid)]=p.stdout.strip() or 'absent'
v['no_tracked_process_running']=all(s=='absent' or ' Z' in s for s in v['tracked_processes'].values())
v['gpu_compute_processes']=cmd('nvidia-smi','--query-compute-apps=gpu_uuid,pid,process_name,used_gpu_memory','--format=csv')
out['hashes']={p.name:sha(p) for p in [Q/'manifest.json',Q/'execution.json',Q/'state.json',Q/'queue-plan.json',Q/'receipt-preflight.json',Q/'authorized_followup_queue.py',Q/'v4_external_timing.py']}
for label,sj in state['jobs'].items():
    p=Q/label; r=read(p/'result.json'); c=out['cases'][label]=dict(status=sj['native_status'],exit_code=sj['exit_code'],timeout=sj['timeout'],process_wall_seconds=sj['wall_seconds'],pid=sj['pid'],gpu_uuid=sj['gpu_uuid'],source_unchanged=r['source_unchanged'])
    c['source_matches_manifest']=all(r['source'][k]==manifest[k] and r['source_after'][k]==manifest[k] for k in ['sha','tree','src_tree','git_status','git_diff_sha256','files_sha256'])
    c['external_manifest_hash_matches']=r['external_manifest_sha256']==sha(Q/'manifest.json')
    c['hashes']={str(z.relative_to(p)):sha(z) for z in [p/'command.json',p/'result.json',*sorted((p/'output').rglob('*.json')),*sorted((p/'output').rglob('*.csv')),*sorted((p/'output').rglob('*.jsonl'))]}
    if label.startswith('v4'):
        n=read(p/'output/result.json')[0]; rows=n['rows']
        captured=r['native_results'][0]
        c['native_receipt_matches_external_result']=all(n[k]==z for k,z in captured.items() if k not in ['last_row','rows_count']) and captured['last_row']==rows[-1] and captured['rows_count']==len(rows)
        c['external_receipt_history_handling']='V4 external receipt stores row count and last row; full native rows are separately hashed'
        c['native_all_finite']=finite(n)
        c['native']={k:z for k,z in n.items() if k not in ['rows','lattice']}
        c['full_series']={k:stats(rows,k) for k in ['cd','cl','p_diff']}
        c['ten_tstar_windows']={str(a):{k:stats([z for z in rows if a<=z['tstar']<=a+10],k) for k in ['cd','cl','p_diff']} for a in range(0,60,10)}
        c['last_window']=dict(tstar=[50,60],metrics={k:stats(rows[-21:],k) for k in ['cd','cl','p_diff']})
        c['health']=dict(rho_min=min(z['rho_min'] for z in rows),rho_max=max(z['rho_max'] for z in rows),max_mach=max(z['max_mach'] for z in rows))
        continue
    case=label.rsplit('-',1)[1]; base=p/'output'/case;n=read(base/'summary.json');plan=read(p/'output/plan.json')
    c['native_receipt_matches_external_result']={k:z for k,z in n.items() if k!='history'}==r['native_summary']
    c['external_receipt_history_handling']='V2 external receipt omits the history path; complete history.csv is separately hashed'
    c['native_all_finite']=finite(n)
    c['native_runner_hash_matches']=plan['runner_sha256']==sha(R/'tests/validation/run_refined_dfg.py')
    sm=plan['source_manifest'];c['native_source_capture_matches']=sm['checkout_revision']==SHA and not sm['checkout_dirty'] and all(sha(R/f)==h==manifest['files_sha256'][f] for f,h in sm['sha256'].items())
    c['native_source_capture_file_count']=len(sm['sha256'])
    c['native']={k:z for k,z in n.items() if k not in ['history','cycles','jet_diagnostic_windows']}; c['comparison']=read(p/'output/comparison.json');c['gates']=plan['gates'];c['reference_gates']=plan['reference_gates'];c['diagnostic_windows']=n['jet_diagnostic_windows']
    rows=[{k:float(z) for k,z in row.items()} for row in csv.DictReader((base/'history.csv').open())];late=[x for x in rows if x['tstar']>=290]
    c['health']=dict(rows=len(rows),tstar=[rows[0]['tstar'],rows[-1]['tstar']],all_finite=finite(rows),rho_min=min(z['rho_min'] for z in rows),rho_max=max(z['rho_max'] for z in rows),max_mach=max(z['max_mach'] for z in rows),mass_first=rows[0]['mass'],mass_last=rows[-1]['mass'],mass_change_fraction=(rows[-1]['mass']-rows[0]['mass'])/rows[0]['mass'],mass_late_range_fraction=span([z['mass'] for z in late])/np.mean([z['mass'] for z in late]))
    if n['cycles']:
        tail=n['cycles'][-5:]; periods=[z['period'] for z in tail]; masses=[z['mass_time_mean'] for z in tail]
        c['periodic_actual']=dict(cycles_total=len(n['cycles']),tail_cycles=len(tail),tail_start=tail[0]['start'],tail_end=tail[-1]['end'],period_min=min(periods),period_max=max(periods),period_mean=float(np.mean(periods)),period_relative_range=span(periods)/np.mean(periods),last_cycle_age=rows[-1]['tstar_force']-tail[-1]['end'],consecutive_max_gap=max(abs(a['end']-b['start']) for a,b in zip(tail,tail[1:])),minimum_lift_amplitude=min(z['cl_amplitude'] for z in tail),extrema_ranges={k+'_'+e:span([z[k+'_'+e] for z in tail]) for k in ['cd','cl','dp'] for e in ['min','max']},mass_relative_range=span(masses)/abs(np.mean(masses)),max_cycle_flux_relative=max(z['flux_relative'] for z in tail),last_cycle={k:z for k,z in tail[-1].items() if not k.endswith('_waveform')})
    cv=[json.loads(x) for x in (base/'cv_history.jsonl').read_text().splitlines()];t=np.array([x['time_root'] for x in cv]);inv=np.array([x['inventory'] for x in cv]);flux=np.array([x['outward_flux'] for x in cv]);j=np.array([x['interval_impulse_root'] for x in cv[1:]]).sum(axis=0);duration=t[-1]-t[0];norm=duration*.5*.01**2*40
    cvout=c['cv']=dict(samples=len(cv),all_finite=finite(cv),tstar=[cv[0]['tstar'],cv[-1]['tstar']],duration_root=duration,coefficient_normalization=norm,raw_body_reaction_impulse_root=j.tolist(),first_preceding_impulse_excluded=True,contours_root=plan['cv_observation'][case+'_contours_root'],mass_qualification='Apparent mass source only: raw Jet mass_into_fluid ledger was not recorded, so mass-source-subtracted closure cannot be assessed',acceptance='offline diagnostics only; no CV pass threshold was defined',strides={})
    for stride in [1,2,4]:
        idx=np.arange(0,len(t),stride);assert idx[-1]==len(t)-1
        integral=np.trapz(flux[idx],t[idx],axis=0); delta=inv[-1]-inv[0];jcv=-delta[:,1:]-integral[:,1:];res=jcv-j
        cvout['strides'][str(stride)]=dict(samples=len(idx),momentum_delta_root=delta[:,1:].tolist(),outward_momentum_integral_root=integral[:,1:].tolist(),cv_reaction_impulse_root=jcv.tolist(),residual_impulse_root=res.tolist(),residual_coefficient=(res/norm).tolist(),apparent_mass_source_root=(delta[:,0]+integral[:,0]).tolist(),apparent_mass_source_fraction_initial_inventory=((delta[:,0]+integral[:,0])/inv[0,:,0]).tolist())
    a=np.array(cvout['strides']['1']['residual_coefficient']);cvout['contour_residual_coefficient_difference']=(a[1]-a[0]).tolist();cvout['coarsening_coefficient_difference']={s:(np.array(cvout['strides'][s]['residual_coefficient'])-a).tolist() for s in ['2','4']}
v['all_six_source_unchanged']=all(c['source_unchanged'] and c['source_matches_manifest'] and c['external_manifest_hash_matches'] for c in out['cases'].values())
v['all_six_native_receipts_match']=all(c['native_receipt_matches_external_result'] and c['native_all_finite'] for c in out['cases'].values())
v['all_v2_native_captures_match']=all(c['native_runner_hash_matches'] and c['native_source_capture_matches'] for k,c in out['cases'].items() if k.startswith('v2'))
out['limitations']=['V1 historical source pin remains blocked','V3 original TG source guard blocked all four dry-runs before CUDA; no TG result or acceptance','Short suite remains 767 passed, 24 failed, 1 skipped; not a release pass','Matched_case_pass remains false because each invocation selected one case; native paired-case comparison branch did not execute','Fixed Jet CV is diagnostic; mass source ledger absent; no physical acceptance','V4 both NOT_STEADY with original ranges/drifts/bands; boundary deviations retained','No formal performance claim: timings include observation/profiling and startup is separately recorded']
dest=Q/'final-physical-digest.json';dest.write_text(json.dumps(out,indent=2,allow_nan=False)+'\n')
print(json.dumps(dict(output=str(dest),bytes=dest.stat().st_size,sha256=sha(dest),verification=v),indent=2))