"""Prospective fixed follow-up; original economic engine in bounded processes."""
import argparse
import asyncio
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
import ctypes
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import threading
import time
from uuid import uuid4

from experiment import Runner,plans
from provider import dump,sha,utc

ROOT=Path(__file__).resolve().parent
PHASE='runtime-followup-v1'
FILES=['runtime_followup.py','experiment.py','provider.py','a2a_harness.py','agent-instructions.txt','requirements.lock.txt','analysis.py','replication_analysis.py','verify_evidence.py','test_research.py','test_evidence.py','test_runtime_followup.py','FOLLOWUP-PROTOCOL.md','RESEARCH-PLAN.md','RESEARCH-JOURNAL.md']


def freeze():
 folder=ROOT/'protocols'/PHASE;folder.mkdir(parents=True,exist_ok=False)
 plan=plans(PHASE,8,12,0,2026090603,['openai'],['hl','hl_blind_l'])
 plan.update(episode_concurrency=2,provider_concurrency=1,worker_deadline_seconds=900,supervised_runtime=True)
 dump(folder/'plan.json',plan);(folder/'source').mkdir()
 for name in FILES:shutil.copy2(ROOT/name,folder/'source'/name)
 dump(folder/'seal.json',{'sealed_at':utc(),'phase':PHASE,'note':'Local prospective hash seal for an adaptively selected follow-up; not external preregistration.','files':{p.relative_to(folder).as_posix():sha(p.read_bytes()) for p in sorted(folder.rglob('*')) if p.is_file()}})
 print(json.dumps({'phase':PHASE,'jobs':len(plan['jobs']),'max_decisions':plan['max_model_decisions'],'seal_sha256':sha((folder/'seal.json').read_bytes())}))


def verify_source():
 protocol=ROOT/'protocols'/PHASE;seal=json.loads((protocol/'seal.json').read_text(encoding='utf-8'))
 for name,digest in seal['files'].items():
  if sha((protocol/name).read_bytes())!=digest:raise RuntimeError('Modified sealed file: '+name)
 for name in ['runtime_followup.py','experiment.py','provider.py','a2a_harness.py','agent-instructions.txt']:
  if (ROOT/name).read_bytes()!=(protocol/'source'/name).read_bytes():raise RuntimeError('Working source changed: '+name)
 return protocol,json.loads((protocol/'plan.json').read_text(encoding='utf-8'))


async def worker(run,job,fixture=False):
 run=Path(run);plan=json.loads((run/'plan.json').read_text(encoding='utf-8'));job=next(j for j in plan['jobs'] if j['id']==job)
 folder=run/'episodes'/job['id'];folder.mkdir(parents=True,exist_ok=False);dump(folder/'job.json',job)
 state={'job':job,'started_at':utc(),'status':'running','fixture':fixture};dump(folder/'status.json',state)
 runner=Runner(run,plan,fixture=fixture,concurrency=1,provider_concurrency=1)
 try:state.update(status='completed',metrics=await runner.market(job,folder))
 except Exception as e:state.update(status='incomplete',error_type=type(e).__name__,error=str(e))
 state['ended_at']=utc();dump(folder/'status.json',state)
 print(json.dumps({'episode':job['id'],'status':state['status']}),flush=True)


def bounded_process(command,directory,deadline,stdout,stderr):
 flags=subprocess.CREATE_NO_WINDOW if os.name=='nt' else 0
 with Path(stdout).open('wb') as output,Path(stderr).open('wb') as error:
  process=subprocess.Popen(command,cwd=directory,stdout=output,stderr=error,creationflags=flags)
  try:code=process.wait(timeout=deadline);return {'pid':process.pid,'exit_code':code,'deadline_exceeded':False}
  except subprocess.TimeoutExpired:
   if os.name=='nt':subprocess.run(['taskkill.exe','/PID',str(process.pid),'/T','/F'],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,check=False)
   else:process.kill()
   process.wait(timeout=30)
   return {'pid':process.pid,'exit_code':process.returncode,'deadline_exceeded':True,'terminated_at':utc()}


def supervise():
 protocol,plan=verify_source();run=ROOT/'private/runs'/(PHASE+'-'+utc().replace(':','').replace('.','')+'-'+str(uuid4())[:8]);run.mkdir(parents=True,exist_ok=False)
 dump(run/'plan.json',plan);shutil.copy2(protocol/'seal.json',run/'protocol-seal.json')
 failures=Counter();lock=threading.Lock();finished=[];started_at=utc()
 kernel=ctypes.WinDLL('kernel32',use_last_error=True) if os.name=='nt' else None
 if kernel:
  kernel.SetThreadExecutionState.argtypes=[ctypes.c_uint];kernel.SetThreadExecutionState.restype=ctypes.c_uint
  previous=kernel.SetThreadExecutionState(0x80000001)
  if not previous:raise RuntimeError('Could not request system execution state')
 else:previous=None
 dump(run/'execution-settings.json',{'started_at':started_at,'supervisor_pid':os.getpid(),'worker_concurrency':2,'provider_slots_per_worker':1,'worker_deadline_seconds':900,'windows_execution_state_flags':'ES_CONTINUOUS | ES_SYSTEM_REQUIRED','previous_execution_state':previous,'persistent_power_policy_changed':False})
 print('RUN_DIRECTORY='+str(run),flush=True)
 def launch(job):
  folder=run/'episodes'/job['id']
  with lock:suspended=any(failures[p]>=2 for p in job['profiles'])
  if suspended:
   folder.mkdir(parents=True,exist_ok=False);dump(folder/'job.json',job);dump(folder/'status.json',{'job':job,'status':'incomplete','fixture':False,'started_at':utc(),'ended_at':utc(),'error_type':'ProfileSuspended','error':'No worker started after two failed adapter records for a profile.'});result={'worker_started':False}
  else:
   logs=run/'worker-logs';logs.mkdir(exist_ok=True)
   result=bounded_process([sys.executable,'-u',str(ROOT/'runtime_followup.py'),'worker','--run',str(run),'--job',job['id']],ROOT,900,logs/(job['id']+'.stdout.txt'),logs/(job['id']+'.stderr.txt'))
   result['worker_started']=True
   if result['deadline_exceeded']:dump(folder/'worker-finalization.json',{**result,'original_status_preserved':True,'note':'Fixed process deadline; no replacement inference or prices.'})
  records=[json.loads(p.read_text(encoding='utf-8')) for p in folder.glob('provider/*/metadata.json')]
  with lock:
   failures.update(m['profile'] for m in records if m['status']!='succeeded')
   finished.append({'episode':job['id'],**result})
   dump(run/'supervisor-progress.json',{'updated_at':utc(),'finished_workers':len(finished),'planned':16,'profile_failed_records':dict(failures)})
   print(json.dumps({'episode':job['id'],'done':len(finished),'of':16,**result}),flush=True)
 try:
  with ThreadPoolExecutor(max_workers=2) as pool:list(pool.map(launch,plan['jobs']))
  dump(run/'completion.json',{'started_at':started_at,'ended_at':utc(),'planned':16,'worker_results':finished,'profile_failed_records':dict(failures),'note':'Worker completion is distinct from valid full-window economic outcomes.'})
 finally:
  if kernel:cleared=kernel.SetThreadExecutionState(0x80000000);dump(run/'power-request-release.json',{'released_at':utc(),'return_value':cleared})
 print('FINISHED='+str(run),flush=True)


if __name__=='__main__':
 ap=argparse.ArgumentParser();ap.add_argument('command',choices=['freeze','run','worker']);ap.add_argument('--run');ap.add_argument('--job');ap.add_argument('--fixture',action='store_true');a=ap.parse_args()
 if a.command=='freeze':freeze()
 elif a.command=='run':supervise()
 else:
  if not a.fixture:verify_source()
  asyncio.run(worker(a.run,a.job,a.fixture))
