#!/usr/bin/env python3
"""Real CLI / generated shell capture. Synthetic metadata cases are explicitly labelled."""
import os, sys, json, hashlib, subprocess, pathlib, datetime, socket, ctypes, signal, time, shutil, re
R = pathlib.Path('/tmp/lead-29428-r1/evidence-capture')
RUN = R / ('run-' + datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%S.%fZ'))
RUN.mkdir()
ctypes.CDLL(None).prctl(36, 1, 0, 0, 0)  # Adopt/reap only descendants spawned by this driver.
SIDES = {'before':('/home/ubuntu/wt/pr-29428-base','edc37c96c810d42761ab510c6a3e9d0a741547dc'), 'after':('/home/ubuntu/wt/pr-29428','b39fd402c62559744f892a2e164212a79e0a6ede')}
C='0.1.6'; L='0.1.5'; H='0.1.7-nightly.20261102.4080'
KEY=hashlib.sha256(b'proof-local\0proof-local\0\0').hexdigest()[:16]
NODE=subprocess.check_output(['node','-p','process.execPath'],text=True,cwd=SIDES['after'][0]).strip()
BIN=RUN/'bin'; BIN.mkdir(); (BIN/'ssh').symlink_to(R/'ssh-stub.py')
(R/'ssh-stub.py').chmod(0o755)
def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='milliseconds')
def sha(p): return hashlib.sha256(pathlib.Path(p).read_bytes()).hexdigest() if pathlib.Path(p).is_file() else 'MISSING'
def status(pid):
    try:
        state=pathlib.Path(f'/proc/{pid}/stat').read_text().split(') ')[1].split()[0]
        return {'pid':pid,'state':state,'live':state!='Z'}
    except FileNotFoundError: return {'pid':pid,'state':'gone','live':False}
def reap(pid):
    try: os.waitpid(pid,os.WNOHANG)
    except ChildProcessError: pass
def env(home):
    return {'HOME':str(home),'PATH':str(BIN)+':'+os.environ['PATH'],'PROOF_HOME':str(home),'PROOF_PATH':str(BIN)+':'+os.environ['PATH'],
      'T3CODE_NO_BROWSER':'1','T3CODE_DEV_SEED':'0','NO_COLOR':'1','FORCE_COLOR':'0','TERM':'dumb','LANG':'C.UTF-8',
      'XDG_CONFIG_HOME':str(home/'.config'),'XDG_CACHE_HOME':str(home/'.cache'),'XDG_DATA_HOME':str(home/'.local/share'),
      'NODE_COMPILE_CACHE':str(home/'.node-cache')}
def freeport():
    with socket.socket() as s: s.bind(('127.0.0.1',0)); return s.getsockname()[1]
class Capture:
    def __init__(self,row,side):
        self.row,self.side=row,side; self.checkout,self.commit=SIDES[side]
        self.root=RUN/f'{row}-{side}'; self.root.mkdir(); self.home=self.root/'home'; self.home.mkdir()
        self.state=self.home/'.t3/ssh-launch'/KEY; self.state.mkdir(parents=True)
        self.runtime=self.home/'.t3/userdata/server-runtime.json'; self.pids=set(); self.n=0
        self.log=(self.root/f'{row}-{side}.log').open('w')
        self.emit('ENVIRONMENT',{'fixture':row,'side':side,'sourceSHA':self.commit,'checkout':self.checkout,'platform':os.uname().sysname+' '+os.uname().machine,
          'harness':'Linux local-SSH stub; actual generated script and built CLI; fresh synthetic HOME; no real SSH or UI',
          'build':'vp run build:bundle plus own web build; archive preinstall wrapper execs actual built bin.mjs; unmodified real L/H release archives',
          'home':str(self.home),'node':subprocess.check_output([NODE,'--version'],text=True).strip(),'stateKey':KEY})
        (self.state/'port').write_text(str(freeport())+'\n')
        self.install(C)
    def emit(self,label,value):
        text=value if isinstance(value,str) else json.dumps(value,sort_keys=True)
        text=re.sub(r'^.*(?:Pairing URL:|App link:|[\u2580-\u259f]).*$', '<GENERATED_PAIRING_PRESENTATION_REDACTED>',text,flags=re.M)
        text=re.sub(r'(?i)(token=|token%3d)[A-Z0-9_-]+',r'\1<REDACTED>',text)
        text=re.sub(r'(?im)^Token: .+$','Token: <GENERATED_PAIRING_TOKEN_REDACTED>',text)
        text=re.sub(r'(?i)(Token:\s*)[A-Z0-9_-]{8,}',r'\1<GENERATED_PAIRING_TOKEN_REDACTED>',text)
        self.log.write(f'{now()} {label} '+text+'\n'); self.log.flush()
    def cmd(self,args,label,expected=None):
        self.emit('COMMAND '+label,args)
        p=subprocess.run(args,env=env(self.home),cwd=self.root,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
        self.emit('EXIT '+label,p.returncode); self.emit('STDOUT '+label,p.stdout); self.emit('STDERR '+label,p.stderr)
        if expected is not None: self.check(label+' expected exit '+str(expected),p.returncode==expected)
        return p
    def check(self,label,ok):
        self.emit('ASSERT '+label,'PASS' if ok else 'FAIL')
        if not ok: raise AssertionError(label)
    def install(self,version):
        d=self.home/'.t3/runtime/versions'/version; d.mkdir(parents=True,exist_ok=True)
        if version==C:
            cli=pathlib.Path(self.checkout)/'apps/server/dist/bin.mjs'
            (d/'t3').write_text('#!/bin/sh\nexec '+NODE+" '"+str(cli)+"' \"$@\"\n"); (d/'t3').chmod(0o755)
            self.emit('ARTIFACT',{'version':version,'kind':'packaging wrapper; complete real built CLI, no helper shim','binSha256':sha(cli),'wrapper':(d/'t3').read_text()})
        else:
            src=R/'releases'/f't3-{version}-linux-x64'
            for child in src.iterdir(): (d/child.name).symlink_to(child,target_is_directory=child.is_dir())
            self.emit('ARTIFACT',{'version':version,'kind':'unmodified release archive','archiveSha256':sha(R/f't3-{version}-linux-x64.tar.gz'),'executableSha256':sha(d/'t3')})
        (d/'.install-complete').write_text(version+'\n')
        self.cmd([str(d/'t3'),'--version'],'version '+version,0)
    def snapshot(self,label):
        values={}
        for name in ['pid','port','managed']:
            p=self.state/name; values[name]=p.read_text().strip() if p.exists() else None
        p=int(values['pid']) if values['pid'] else None
        if p: self.pids.add(p); values['process']=status(p)
        try: values['runtime']=json.loads(self.runtime.read_text())
        except (FileNotFoundError,json.JSONDecodeError): values['runtimeRaw']=self.runtime.read_text() if self.runtime.exists() else 'MISSING'
        values['runnerSha256']=sha(self.state/'run-t3.sh')
        self.emit(label,values)
        return values
    def connect(self,version,label):
        self.n+=1; d=self.root/f'connect-{self.n:02d}'; d.mkdir()
        beforehash=sha(self.state/'run-t3.sh')
        self.emit('STEP '+label,{'archiveVersion':version,'previousRunnerSha256':beforehash})
        self.snapshot('PRE_CONNECT')
        e=env(self.home); e['PROOF_CAPTURE']=str(d)
        args=[NODE,str(R/'launch.mjs'),self.checkout,version]
        self.emit('COMMAND launchOrReuseRemoteServer',args)
        p=subprocess.run(args,env=e,cwd=self.root,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
        self.emit('LAUNCH_EXIT',p.returncode); self.emit('LAUNCH_STDOUT',p.stdout); self.emit('LAUNCH_STDERR',p.stderr)
        for name in ['ssh.log','remote.stdout','remote.stderr']:
            if (d/name).exists(): self.emit(name,(d/name).read_text())
        post=self.snapshot('POST_CONNECT')
        proposed=sha(self.state/'run-t3.sh')
        self.emit('RUNNER_COMPARISON',{'previous':beforehash,'proposed':proposed,'changed':beforehash!=proposed})
        self.reap_all()
        return p,post
    def reap_all(self):
        for pid in self.pids: reap(pid)
    def helper(self,name,contents,pid=4242,expected=1,nightly=False,path=None):
        fixture=self.root/f'helper-{name}.json' if path is None else path
        if contents is not None: fixture.write_text(contents if isinstance(contents,str) else json.dumps(contents))
        self.emit('HELPER_FIXTURE '+name,{'path':str(fixture),'contents':contents,'pidArgument':pid,'synthetic':True})
        cli=R/'nightly-source/apps/server/dist/bin.mjs' if nightly else pathlib.Path(self.checkout)/'apps/server/dist/bin.mjs'
        if nightly:
            self.emit('NIGHTLY_VERSION_ONLY',{'source':self.commit,'packageVersion':'0.1.6-nightly.20261102.4080','binSha256':sha(cli)})
            self.cmd([NODE,str(cli),'--version'],'nightly version',0)
        return self.cmd([NODE,str(cli),'__ssh-helper','keep-server',str(fixture),str(pid)],'helper '+name,expected)
    def cleanup(self):
        self.snapshot('PRE_CLEANUP')
        for pid in self.pids:
            if status(pid)['live']:
                self.emit('CLEANUP SIGCONT/SIGTERM',pid)
                try:
                    os.kill(pid,signal.SIGCONT); os.kill(pid,signal.SIGTERM)
                except ProcessLookupError: pass
        deadline=time.monotonic()+15
        while any(status(p)['live'] for p in self.pids) and time.monotonic()<deadline:
            self.reap_all(); time.sleep(.05)
        for pid in self.pids:
            if status(pid)['live']:
                self.emit('CLEANUP SIGKILL tracked child',pid); os.kill(pid,signal.SIGKILL)
            try: os.waitpid(pid,0)
            except ChildProcessError: pass
            self.emit('CLEANUP_FINAL',status(pid))
        # Preserve raw server output but redact only generated one-time pairing links from public logs.
        serverlog=self.state/'server.log'
        if serverlog.exists():
            import re
            text=serverlog.read_text(errors='replace')
            text=re.sub(r'https?://[^\s"<>]+(?:pair|token|credential)[^\s"<>]*','<GENERATED_PAIRING_URL_REDACTED>',text)
            self.emit('SERVER_LOG',text)
        self.emit('END',self.row+' '+self.side); self.log.close()
def launchrow(c):
    row,side=c.row,c.side
    initial=H if row=='A01' else L if row=='A03' else C
    if initial!=C: c.install(initial)
    p,start=c.connect(initial,'1 cold launch real initial server')
    c.check('initial launch successful',p.returncode==0)
    old=int(start['pid']); initialRuntime=start['runtime']; c.check('initial version',initialRuntime['serverVersion']==initial)
    if row in ['A02','A06','A07','A08','A09','A11']:
        with (c.state/'run-t3.sh').open('a') as f: f.write('\n')
        c.emit('HARNESS_MUTATION','Append harmless newline to remembered runner; force bytewise RUNNER_CHANGED=1')
    if row=='A06':
        q=subprocess.Popen([NODE,'-e','process.exit(0)'],env=env(c.home)); dead=q.pid; q.wait()
        c.emit('DEAD_TRACKED_FIXTURE_PID',status(dead)); c.check('Q demonstrably dead',not status(dead)['live'])
        j=dict(initialRuntime); j['pid']=dead; c.runtime.write_text(json.dumps(j)); c.emit('HARNESS_MUTATION',{'runtimePid':dead,'managedPid':old})
        if side=='after': c.helper('pid-mismatch',j,pid=old)
    if row=='A07':
        c.runtime.rename(c.runtime.with_name('server-runtime.preserved.json')); c.emit('HARNESS_MUTATION','Move runtime to preserved sibling; runtime path missing')
        if side=='after': c.helper('missing',None,path=c.root/'never-created.json')
    if row=='A08':
        c.runtime.write_text('{'); c.emit('HARNESS_MUTATION','Replace runtime contents with malformed {')
        if side=='after':
            variants={'empty':'','malformed':'{','null':'null','array':'[]','missing-pid':{'serverVersion':C},'string-pid':{'pid':'4242','serverVersion':C},'missing-version':{'pid':4242},'number-version':{'pid':4242,'serverVersion':6}}
            for name,value in variants.items(): c.helper(name,value)
    if row=='A09':
        j=dict(initialRuntime); j['serverVersion']='latest'; c.runtime.write_text(json.dumps(j)); c.emit('HARNESS_MUTATION',{'serverVersion':'latest'})
        if side=='after': c.helper('non-exact',j,pid=old)
    if row=='A02' and side=='after': c.helper('build-metadata',{'pid':old,'serverVersion':'0.1.6+proof'},pid=old,expected=0)
    if row=='A11':
        os.kill(old,signal.SIGSTOP); c.emit('STEP SIGSTOP only tracked server',status(old))
    p,end=c.connect(C,'2 reconnect connecting client C')
    c.emit('OLD_PID_LIVENESS',status(old))
    if row in ['A01','A02'] and side=='after':
        c.check('old PID retained and live',end['pid']==str(old) and status(old)['live'])
        c.check('runtime version and startedAt retained',end['runtime']['serverVersion']==initial and end['runtime']['startedAt']==initialRuntime['startedAt'])
        c.check('returns managed same actual port',p.returncode==0 and end['port']==start['port'] and end['managed']=='managed')
    elif row=='A11':
        if side=='after':
            c.check('readiness error leaves stopped PID owned',p.returncode!=0 and 'server was left running' in p.stderr and end['pid']==str(old) and status(old)['live'])
            c.check('runtime unchanged',end['runtime']==initialRuntime)
        else:
            c.emit('BASELINE_SIGSTOP_OBSERVATION','Baseline sends SIGTERM to stopped old server and attempts replacement; SIGTERM stays pending until SIGCONT. The same scripted interaction is captured without requiring baseline preservation.')
        os.kill(old,signal.SIGCONT); c.emit('STEP SIGCONT tracked server',status(old))
    else:
        c.check('old PID stopped',not status(old)['live'])
        if p.returncode==0:
            c.check('distinct C server managed',end['pid']!=str(old) and end['runtime']['serverVersion']==C and end['managed']=='managed')
        elif row=='A01' and side=='before':
            c.emit('OBSERVED_BASELINE_FAILURE','C replacement startup failed after newer H database migration; real error recorded above')
        else: c.check('replacement launch successful',False)
def runone(row,side):
    c=Capture(row,side)
    try:
        if row in ['A04','A05','A10']:
            if row=='A04': c.helper('nightly-below-stable',{'pid':4242,'serverVersion':'0.1.6-nightly.20261102.4080'},expected=1 if side=='after' else None)
            if row=='A05': c.helper('stable-above-nightly',{'pid':4242,'serverVersion':C},expected=0 if side=='after' else None,nightly=side=='after')
            if row=='A10':
                d=c.root/'runtime-directory'; d.mkdir(); p=c.helper('read-directory',None,path=d,expected=None); c.check('directory read nonzero',p.returncode!=0)
            if side=='before': c.emit('BASELINE_CONTEXT','keep-server command absent; actual base CLI invocation is recorded, not an equivalent baseline guard')
        else: launchrow(c)
        c.emit('ROW_RESULT','CAPTURED')
    except Exception as e:
        c.emit('ROW_RESULT','UNVERIFIED '+repr(e)); raise
    finally: c.cleanup()
if __name__=='__main__':
    print(str(RUN),flush=True)
    rows=sys.argv[1:] or [f'A{i:02d}' for i in range(1,12)]
    errors=[]
    for row in rows:
        for side in ['before','after']:
            print(now(),row,side,flush=True)
            try: runone(row,side)
            except Exception as e: errors.append((row,side,repr(e))); print('ERROR',repr(e),flush=True)
    (RUN/'errors.json').write_text(json.dumps(errors,indent=2)+'\n')
    print('ERRORS',errors,flush=True)