#!/usr/bin/env python3
"""Bounded T01 confirmation harness; only disposable output directories are written.
Uses real source preflight and runner execution, never journal-supplied executable code.
ACTION3 R01 integrates a test-only source-frame stall for --timeout-only. The fault
uses a real subprocess timeout; normal execution and final resume are uninjected.
No production/runtime certification or semantic-equivalence proof is implied.
"""
from __future__ import annotations
import argparse,copy,hashlib,importlib.util,json,math,os,runpy,shutil,subprocess,sys,time,types
sys.dont_write_bytecode = True
from pathlib import Path
from datetime import datetime,timezone

def sha(p):return hashlib.sha256(Path(p).read_bytes()).hexdigest()
def write(p,d):Path(p).write_text(json.dumps(d,indent=2)+'\n')

# ACTION3 R01: test-only fault injection, never a candidate-source patch.
# The real selected source frame is held before its first statement. The parent
# waits for that handshake before starting a real subprocess timeout. Startup
# speed and the runtime of unrelated commands therefore cannot select the fault.

def positive_seconds(value):
 value=float(value)
 if not math.isfinite(value) or value <= 0:
  raise argparse.ArgumentTypeError('timing must be finite and positive')
 return value

def nonnegative_seconds(value):
 value=float(value)
 if not math.isfinite(value) or value < 0:
  raise argparse.ArgumentTypeError('delay must be finite and nonnegative')
 return value

def wait_for_source_frame(proc,ready,target,limit):
 deadline=time.monotonic()+limit
 while not ready.exists():
  code=proc.poll()
  if code is not None:
   raise RuntimeError('fault child exited before source-frame handshake: '+str(code))
  if time.monotonic() >= deadline:
   raise RuntimeError('source-frame handshake deadline exceeded; NOT a verified timeout')
  time.sleep(0.01)
 event=json.loads(ready.read_text())
 if (event.get('kind')!='ACTION3_SOURCE_FRAME_STALL' or
     event.get('pid')!=proc.pid or event.get('target')!=str(target) or
     event.get('target_sha256')!=sha(target)):
  raise RuntimeError('invalid source-frame handshake')
 return event

def run_with_source_stall(command,ready,timeout_seconds,ready_seconds,startup_delay,**kwargs):
 """Run actual source through a test-only entry-frame gate; raise real timeout."""
 if not isinstance(command,list) or len(command)<3 or command[1]!='-B':
  raise ValueError('unexpected source command shape')
 target=Path(command[2]).resolve()
 if ready.exists():
  raise ValueError('refusing existing fault evidence')
 gated=[sys.executable,'-B',str(Path(__file__).resolve()),'--action3-gated-child',
        str(ready),str(startup_delay),str(target),*command[3:]]
 kwargs.pop('timeout',None)  # Start fault budget ONLY after the handshake.
 proc=subprocess.Popen(gated,**kwargs)
 try:
  wait_for_source_frame(proc,ready,target,ready_seconds)
  # This is the real subprocess wait, not a fabricated exception or exit code.
  code=proc.wait(timeout=timeout_seconds)
  return subprocess.CompletedProcess(command,code)
 finally:
  if proc.poll() is None:
   proc.kill()
  proc.wait()

def gated_child(argv):
 ready=Path(argv[0]);delay=nonnegative_seconds(argv[1]);target=Path(argv[2]).resolve()
 if ready.exists():
  raise ValueError('ready path already exists')
 time.sleep(delay)  # Optional explicit slow-start regression fixture.
 sys.argv=[str(target),*argv[3:]];sys.path.insert(0,str(target.parent))
 def at_source(frame,event,arg):
  if event=='call' and frame.f_code.co_filename==str(target):
   body={'kind':'ACTION3_SOURCE_FRAME_STALL','pid':os.getpid(),
         'target':str(target),'target_sha256':sha(target),
         'phase':'before_first_statement','test_only':True}
   temp=ready.with_suffix('.pending');write(temp,body);temp.replace(ready)
   print('ACTION3 TEST-ONLY STALL: '+str(target),flush=True)
   while True:time.sleep(3600)  # Parent kills this process on its bounded timeout.
  return at_source
 sys.settrace(at_source)
 try:runpy.run_path(str(target),run_name='__main__')
 finally:sys.settrace(None)
 raise RuntimeError('target frame was never intercepted')

def subprocess_proxy(run):
 return types.SimpleNamespace(**dict(vars(subprocess),run=run))

def fault_runner(argv):
 script=Path(argv[0]).resolve();index=int(argv[1]);ready=Path(argv[2]).resolve()
 budget=positive_seconds(argv[3]);startup=positive_seconds(argv[4]);delay=nonnegative_seconds(argv[5])
 sys.path.insert(0,str(script.parent))
 spec=importlib.util.spec_from_file_location('action3_frozen_runner',script)
 module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
 plan=module.command_plan()
 if not 1<=index<=len(plan):raise ValueError('fault index outside actual plan')
 target_name=plan[index-1][0]
 if sum(name==target_name for name,_ in plan)!=1:raise ValueError('ambiguous fault target')
 injected=[]
 def controlled_run(command,**kwargs):
  if len(command)>=3 and Path(command[2]).name==target_name:
   if injected:raise RuntimeError('duplicate injection attempt')
   injected.append(target_name)
   return run_with_source_stall(command,ready,budget,startup,delay,**kwargs)
  return subprocess.run(command,**kwargs)
 # Change only this imported runner's subprocess binding for this fault run.
 # Do not patch the global subprocess module, source files, plan, hashes or receipts.
 module.subprocess=subprocess_proxy(controlled_run)
 sys.argv=[str(script),*argv[6:]]
 code=module.main()
 if not injected:raise RuntimeError('requested fault was not reached')
 return code

def validate_timeout_checkpoint(seed,journal,failure,index):
 """Derived state checks: no completed work may be credited to a failed attempt."""
 checks=[
  (failure.get('result')=='FAIL','missing failed reproduction receipt'),
  (journal.get('commands')==seed.get('commands'),'verified checkpoint advanced or changed'),
  (len(journal.get('attempts',[]))==len(seed.get('attempts',[]))+1,'unexpected attempt count'),
  (journal.get('state')=='FAILED_OR_INCOMPLETE','journal not marked failed/incomplete'),
 ]
 for passed,reason in checks:
  if not passed:raise AssertionError(reason)
 attempt=journal['attempts'][-1]
 if (attempt.get('index')!=index or attempt.get('exit_code')!=124 or
     attempt.get('state')!='FAILED_OR_INCOMPLETE'):
  raise AssertionError('requested command did not record a genuine incomplete timeout')
 return attempt

def main():
 if sys.flags.optimize:raise RuntimeError('Run without -O / PYTHONOPTIMIZE; assertions must be enabled')
 p=argparse.ArgumentParser(description=__doc__);p.add_argument('--candidate',type=Path,required=True);p.add_argument('--archive',type=Path,required=True);p.add_argument('--out',type=Path,required=True);p.add_argument('--actor',required=True);p.add_argument('--session',required=True);p.add_argument('--timeout-only',action='store_true');p.add_argument('--fault-at',type=int,default=3);p.add_argument('--fault-timeout',type=positive_seconds,default=0.2);p.add_argument('--fault-ready-timeout',type=positive_seconds,default=30.0);p.add_argument('--fault-startup-delay',type=nonnegative_seconds,default=0.0);p.add_argument('--runner-timeout',type=positive_seconds,default=600.0);a=p.parse_args()
 if not 2<=a.fault_at<=12:p.error('--fault-at must be between 2 and 12')
 C=a.candidate.resolve();A=a.archive.resolve();O=a.out.resolve()
 if O.exists():raise ValueError('Output must be new; preserve previous evidence')
 if O==C or O.is_relative_to(C) or C.is_relative_to(O):raise ValueError('Source/output must be disjoint')
 O.mkdir(parents=True);results=[];run=O/'legitimate'
 initial={q.relative_to(C).as_posix():sha(q) for q in C.rglob('*') if q.is_file()}
 script=C/'runner/verify_review_packet.py'
 def invoke(label,out,extra=(),actor=None,archive=None,expected=0,fault=False):
  log=O/(label+'.log');cmd=[sys.executable,'-B',str(script),'--actor',actor or a.actor,'--session',a.session,'--out',str(out),'--predecessor-archive',str(archive or A),*extra]
  if fault:
   cmd=[sys.executable,'-B',str(Path(__file__).resolve()),'--action3-fault-runner',str(script),str(a.fault_at),str(O/'FAULT_READY.json'),str(a.fault_timeout),str(a.fault_ready_timeout),str(a.fault_startup_delay),*cmd[3:]]
  start=time.monotonic()
  with log.open('wb') as f:r=subprocess.run(cmd,stdout=f,stderr=subprocess.STDOUT,timeout=a.runner_timeout)
  record={'id':label,'exit_code':r.returncode,'expected_exit_code':expected,'pass':r.returncode==expected,'log':log.name,'log_sha256':sha(log),'seconds':round(time.monotonic()-start,3)}
  results.append(record);write(O/'CHECKS.json',results);print(json.dumps(record),flush=True)
  if not record['pass']:raise AssertionError(label+' returned unexpected exit code')
  return log
 if a.timeout_only:
  checkpoint=a.fault_at-1
  invoke('pause_after_'+str(checkpoint),run,['--stop-after',str(checkpoint)],expected=75)
  seed=json.loads((run/'COMMAND_PROGRESS.json').read_text())
  assert len(seed['commands'])==checkpoint
  invoke('forced_timeout',run,['--resume'],expected=1,fault=True)
  failure_path=run/'REVIEWER_REPRODUCTION_RECEIPT.json'
  failure=json.loads(failure_path.read_text())
  journal=json.loads((run/'COMMAND_PROGRESS.json').read_text())
  # A failed attempt must not advance the verified checkpoint or invent a PASS.
  failed_attempt=validate_timeout_checkpoint(seed,journal,failure,a.fault_at)
  ready=json.loads((O/'FAULT_READY.json').read_text())
  target=Path(ready['target']);assert target.name==seed['plan'][a.fault_at-1][0]
  assert ready['target_sha256']==sha(C/'runner'/target.name)
  failure_bytes=failure_path.read_bytes();failure_hash=sha(failure_path)
  invoke('resume_after_timeout',run,['--resume'],expected=0)
  receipt=json.loads(failure_path.read_text())
  assert (run/'receipt_history'/f'{failure_hash}.json').read_bytes()==failure_bytes
  journal=json.loads((run/'COMMAND_PROGRESS.json').read_text())
  assert journal['commands'][:checkpoint]==seed['commands']
  assert failed_attempt in journal['attempts']
  assert len(journal['attempts'])==len(seed['plan'])+1
  assert [row['index'] for row in journal['commands']]==list(range(1,len(seed['plan'])+1))
  assert receipt['resumptions'][-1]['completed_commands_restored']==checkpoint
  results.append({'id':'failed_attempt_and_receipt_preserved','pass':True,
                  'failed_receipt_sha256':failure_hash,'fault_at_command':a.fault_at,
                  'completed_commands_restored':checkpoint,
                  'fault_mode':'test-only source-frame stall; real subprocess timeout',
                  'fault_timeout_seconds':a.fault_timeout,
                  'startup_handshake_timeout_seconds':a.fault_ready_timeout,
                  'configured_startup_delay_seconds':a.fault_startup_delay,
                  'fault_ready_sha256':sha(O/'FAULT_READY.json'),
                  'resume_mode':'unmodified real runner; no injection'})
 else:
  invoke('pause_after_9',run,['--stop-after','9'],expected=75)
  seed=json.loads((run/'COMMAND_PROGRESS.json').read_text());assert len(seed['commands'])==9
  # These are not copies of a real user edit. They intentionally forge journals in disposable fixtures.
  for kind in ('replace','add','delete'):
   dest=O/('tampered_'+kind);shutil.copytree(run,dest)
   journal=copy.deepcopy(seed);target='runner/t01_added_fixture.py' if kind=='add' else 'runner/test_hf03.py'
   if kind=='delete':delta={'path':target,'deleted':True}
   else:
    artifact=dest/'forged_fixture.py';artifact.write_text("raise RuntimeError('T01 FORGED FIXTURE MUST NOT EXECUTE')\n")
    delta={'path':target,'artifact':artifact.name,'sha256':sha(artifact)}
   journal['commands'][-1]['delta'].append(delta);write(dest/'COMMAND_PROGRESS.json',journal)
   before=len(journal['attempts']);log=invoke('reject_'+kind,dest,['--resume'],expected=2)
   assert 'delta outside writable evidence scope: '+target in log.read_text()
   after=json.loads((dest/'COMMAND_PROGRESS.json').read_text());assert len(after['attempts'])==before
   results.append({'id':kind+'_rejected_before_next_execution','pass':True,'target':target})
  wrong=O/'wrong_actor';shutil.copytree(run,wrong);invoke('wrong_actor',wrong,['--resume'],actor=a.actor+'-OTHER',expected=2)
  wrong=O/'wrong_archive';shutil.copytree(run,wrong);bad=O/'incorrect_archive.zip';bad.write_bytes(b'wrong-archive-fixture');invoke('wrong_archive',wrong,['--resume'],archive=bad,expected=2)
  # Valid restart must discard an unrecorded mutation, then execute only remaining commands.
  (run/'isolated-candidate/runner/unrecorded_interruption.py').write_text('# abandoned scratch mutation\n')
  invoke('legitimate_resume',run,['--resume'],expected=0)
  assert not (run/'isolated-candidate/runner/unrecorded_interruption.py').exists()
  receipt=json.loads((run/'REVIEWER_REPRODUCTION_RECEIPT.json').read_text());journal=json.loads((run/'COMMAND_PROGRESS.json').read_text())
  assert [x['index'] for x in journal['commands']]==list(range(1,13)) and len(journal['attempts'])==12
  assert receipt['resumptions'][-1]['completed_commands_restored']==9
  results.append({'id':'exactly_remaining_commands_executed','pass':True,'restored':9,'executed_after_resume':[10,11,12]})
 assert receipt['result']=='PASS' and receipt['recounted_tests']['failed']==receipt['recounted_tests']['not_run']==0
 exact=[]
 for row in receipt['result_comparisons']:
  output=run/'reports'/Path(row['path']).name
  equal=output.read_bytes()==(C/row['path']).read_bytes();assert equal
  exact.append({'path':row['path'],'sha256':sha(output),'byte_identical':equal,'recounted':row['recounted']})
 final={q.relative_to(C).as_posix():sha(q) for q in C.rglob('*') if q.is_file()};assert initial==final
 write(O/'CHECKS.json',results)
 report={'kind':'T01_FORCED_INTERRUPTION_CONFIRMATION' if a.timeout_only else 'T01_REAL_RUNNER_RESUME_CONFIRMATION','result':'PASS','actor':a.actor,'session':a.session,'at_utc':datetime.now(timezone.utc).isoformat(),'source_files':len(initial),'source_unchanged':True,'state_sha256':sha(C/'VPL_V5_CURRENT_STATE.json'),'reproduction_receipt_sha256':sha(run/'REVIEWER_REPRODUCTION_RECEIPT.json'),'commands':len(receipt['commands']),'recounted':receipt['recounted_tests'],'reports':exact,'checks':results,'boundary':'Executing actor evidence only. Not independent reviewer acceptance, authenticated identity, provider or production certification.'}
 write(O/'SUMMARY.json',report);print(json.dumps({'result':report['result'],'recounted':report['recounted'],'reports':len(exact)}),flush=True)
if __name__=='__main__':
 if len(sys.argv)>1 and sys.argv[1]=='--action3-gated-child':gated_child(sys.argv[2:])
 elif len(sys.argv)>1 and sys.argv[1]=='--action3-fault-runner':sys.exit(fault_runner(sys.argv[2:]))
 else:main()