#!/usr/bin/env python3
"""Read-only checkout evidence driver. All writes stay in the supplied scratch run.
Uses real CLI processes; SQLite observations are independent short-lived children.
Each invocation creates a fresh timestamped run. No live T3 state is opened.
"""
import argparse, datetime, hashlib, json, os, pathlib, platform, selectors, shlex, shutil, signal, socket, subprocess, sys, time, traceback, uuid
ROOT=pathlib.Path('/tmp/lead-28256-r1/evidence-capture')
NODE='/home/ubuntu/.local/share/vite-plus/js_runtime/node/26.12.1/bin/node'
SCRIPT=ROOT/'capture.py'
PROBE=ROOT/'probe-lock.mjs'
HELPER=ROOT/'helper-no-create.mjs'
ACTIVE=[]
ENV=os.environ.copy()
REMOVED=sorted(k for k in ENV if k.startswith(('T3CODE_','T3_SERVICE_','VITE_')))
for k in REMOVED: ENV.pop(k)
ENV.update({'NO_COLOR':'1','FORCE_COLOR':'0','TZ':'UTC'})
TRANSCRIPT=None

def redact(s):
    import re
    s=re.sub(r'((?:https?|wss?)://[^\s<>"\']+)',lambda m: re.sub(r'([?&#](?:token|secret|code|auth|credential|pairingToken|bootstrapToken)=)[^&#\s]+',r'\1<redacted>',m.group(0),flags=re.I),s)
    s=re.sub(r'((?:pairing|bootstrap|auth|access)[_-]?token["\']?\s*[:=]\s*["\']?)[A-Za-z0-9_./+=-]{16,}',r'\1<redacted>',s,flags=re.I)
    return s

def log(s):
    s=redact(str(s))
    print(s,flush=True)
    if TRANSCRIPT: TRANSCRIPT.write(s+'\n'); TRANSCRIPT.flush()

def command(argv,cwd=None):
    log('$ '+('cd '+shlex.quote(str(cwd))+' && ' if cwd else '')+shlex.join(map(str,argv)))
    p=subprocess.run(list(map(str,argv)),cwd=cwd,env=ENV,text=True,capture_output=True)
    for line in p.stdout.splitlines(): log('[command stdout] '+line)
    for line in p.stderr.splitlines(): log('[command stderr] '+line)
    log(f'command wait: exit={p.returncode}')
    return p

def state(base,label,project=None):
    log('STATE '+label)
    r=base/'userdata/server-runtime.json'; l=base/'userdata/server-owner.sqlite'
    if r.exists():
        data=r.read_bytes()
        log('R path='+str(r)+' sha256='+hashlib.sha256(data).hexdigest())
        log('R contents BEGIN\n'+data.decode(errors='replace')+'R contents END')
    else: log('R absent: '+str(r))
    try:
        st=l.stat()
        meta={'exists':True,'path':str(l),'dev':st.st_dev,'inode':st.st_ino,'mode':oct(st.st_mode&0o7777),'uid':st.st_uid,'gid':st.st_gid,'size':st.st_size}
        try:
            b=l.read_bytes();meta['sha256']=hashlib.sha256(b).hexdigest();meta['first64bytes_hex']=b[:64].hex()
        except OSError as e:meta['sha256']='unavailable: '+str(e)
        log('L '+json.dumps(meta,sort_keys=True))
    except FileNotFoundError:log('L absent: '+str(l))
    if project:log('target cwd exists='+str(project.exists())+' path='+str(project))

def probe(base):
    log('INDEPENDENT LOCK PROBE (process closes before next target launch)')
    return command([NODE,PROBE,base/'userdata/server-owner.sqlite'])

def freeport():
    with socket.socket() as s:s.bind(('127.0.0.1',0));return s.getsockname()[1]

def live(p,label):
    rc=p.poll()
    try:os.kill(p.pid,0);alive=True
    except ProcessLookupError:alive=False
    log(f'{label}: pid={p.pid} signal-0-alive={alive} poll={rc}')
    return alive and rc is None

def stop(p,sig,label):
    if p.poll() is None:
        log(f'CLEANUP {label}: send {signal.Signals(sig).name} to spawned pid={p.pid}')
        p.send_signal(sig)
    else:log(f'CLEANUP {label}: already exited pid={p.pid}')
    try:rc=p.wait(timeout=25)
    except subprocess.TimeoutExpired:
        log(f'CLEANUP {label}: escalated SIGKILL pid={p.pid}')
        p.kill();rc=p.wait(timeout=10)
    log(f'WAIT {label}: pid={p.pid} returncode={rc} '+(f'signal={signal.Signals(-rc).name}' if rc<0 else f'exit={rc}'))
    return rc

class Target:
    def __init__(self,checkout,base,project,port,run,label):
        self.checkout=checkout;self.base=base;self.project=project;self.port=port;self.label=label;self.run=run
        self.out='';self.err='';self.sel=selectors.DefaultSelector();self.handles=[]
        argv=[NODE,checkout/'apps/server/src/bin.ts','start','--mode','web','--base-dir',base,'--port',port,'--host','127.0.0.1','--no-browser',project]
        log('TARGET '+label+' argv='+shlex.join(map(str,argv)))
        self.p=subprocess.Popen(list(map(str,argv)),cwd=checkout/'apps/server',env=ENV,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
        ACTIVE.append(self.p)
        log(f'SPAWN {label}: tracked pid={self.p.pid} euid={os.geteuid()} pidns={os.readlink("/proc/self/ns/pid")}')
        for name,stream in [('stdout',self.p.stdout),('stderr',self.p.stderr)]:
            os.set_blocking(stream.fileno(),False)
            h=open(run/(label+'.'+name+'.log'),'w');self.handles.append(h)
            self.sel.register(stream,selectors.EVENT_READ,(name,h))
    def pump(self,timeout=.2):
        for key,_ in self.sel.select(timeout):
            b=os.read(key.fileobj.fileno(),65536)
            if not b:self.sel.unregister(key.fileobj);continue
            s=redact(b.decode(errors='replace'));name,h=key.data;h.write(s);h.flush()
            if name=='stdout':self.out+=s
            else:self.err+=s
            for line in s.splitlines():log(f'[{self.label} {name}] '+line)
    def await_result(self,limit=120):
        until=time.monotonic()+limit
        while time.monotonic()<until:
            self.pump()
            r=self.base/'userdata/server-runtime.json'
            if self.p.poll() is not None:
                self.drain();log(f'TARGET {self.label} wait: exit={self.p.returncode}');return 'exited'
            if 'Listening on ' in self.out+self.err and r.exists():
                try:d=json.loads(r.read_text())
                except (ValueError,OSError):continue
                if d.get('pid')==self.p.pid and d.get('port')==self.port and d.get('ownerId'):
                    log(f'READINESS {self.label}: alive + Listening line + new valid R matches pid/port with ownerId')
                    state(self.base,self.label+' ready',self.project)
                    q=probe(self.base)
                    if '"classification":"busy"' not in q.stdout:raise RuntimeError('readiness lock not busy')
                    live(self.p,self.label+' ready');return 'started'
        raise RuntimeError('target readiness/exit deadline exceeded '+self.label)
    def drain(self):
        while self.sel.get_map():self.pump(.1)
    def cleanup(self,sig=signal.SIGTERM):
        rc=stop(self.p,sig,self.label);self.drain()
        for h in self.handles:h.close()
        return rc

def seed(base,pid,port,legacy=False):
    d={'version':1,'pid':pid,'port':port,'origin':f'http://127.0.0.1:{port}','startedAt':datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='milliseconds').replace('+00:00','Z')}
    if not legacy:d['ownerId']=str(uuid.uuid4())
    (base/'userdata').mkdir(parents=True,exist_ok=True)
    (base/'userdata/server-runtime.json').write_text(json.dumps(d,separators=(',',':'))+'\n')
    log('SEED '+('legacy-format integration input (no ownerId)' if legacy else 'lock-aware synthetic integration input; unrelated live child PID; no real server associated with R'))

def header(checkout,row,run,side):
    log(f'ROW {row} SIDE {side} run={run} checkout={checkout}')
    command(['git','-C',checkout,'rev-parse','HEAD'])
    command([NODE,'--version'])
    log(f'ENV machine=fs-20 hostname={socket.gethostname()} platform={platform.platform()} node={NODE} euid={os.geteuid()} egid={os.getegid()} host_uid={os.getuid()} pid={os.getpid()} pidns={os.readlink("/proc/self/ns/pid")}')
    log('ENV removed ambient keys='+json.dumps(REMOVED)+'; overrides='+json.dumps({'NO_COLOR':'1','FORCE_COLOR':'0','TZ':'UTC'})+'; HOME unchanged; explicit isolated --base-dir; no dev URL')
    command(['df','-BG','--output=avail','/'])

def ordinary(checkout,row,run):
    base=run/'base';base.mkdir();project=run/'new-project';port=freeport()
    initial=None;target=None;child=None
    try:
        if row=='F07':
            state(base,'before A',run/'owner-project');probe(base)
            initial=Target(checkout,base,run/'owner-project',freeport(),run,'A')
            if initial.await_result()!='started':raise RuntimeError('owner A did not start')
            original=(base/'userdata/server-runtime.json').read_bytes()
        else:
            child=subprocess.Popen(['/bin/sleep','600'],env=ENV);ACTIVE.append(child)
            log(f'SPAWN unrelated child argv=/bin/sleep 600 tracked pid={child.pid}')
            live(child,'unrelated fixture child')
            seed(base,child.pid,freeport(),legacy=row=='F06')
            lock=base/'userdata/server-owner.sqlite'
            if row=='F10':
                lock.write_bytes(b'This is deliberately NOT a SQLite database. Fixture F10.\n'*8)
                log('CORRUPT L seeded with non-SQLite bytes; no lock owner')
            if row=='F11':
                lock.write_bytes(b'')
                log('PERMISSION L seed sha256='+hashlib.sha256(lock.read_bytes()).hexdigest()+' (before chmod 000)')
                lock.chmod(0)
                log('PERMISSION L chmod 000; parent writable='+str(os.access(lock.parent,os.W_OK)))
                try:lock.open('rb').close();raise RuntimeError('chmod 000 unexpectedly readable')
                except PermissionError as e:log('VERIFIED real non-root read denied: '+str(e))
        state(base,'immediately before target B',project);probe(base)
        if row=='F09' and 'pr-28256-base' not in str(checkout):
            log('F09 EXTRA helper-level no-create check before real CLI')
            command([NODE,HELPER,checkout,base/'userdata/server-runtime.json'],cwd=checkout/'apps/server')
            state(base,'after helper; before CLI',project)
        if child:live(child,'unrelated child before CLI')
        target=Target(checkout,base,project,port,run,'B')
        result=target.await_result()
        state(base,'after target result; before cleanup',project);probe(base)
        if initial:
            log('R byte-identical to A publication='+str(original==(base/'userdata/server-runtime.json').read_bytes()))
            live(initial.p,'A remains alive after B')
        if child:live(child,'unrelated child after CLI')
        if result=='started':
            log('OUTCOME target B STARTED; long-lived process, no exit claimed at readiness')
        else:log(f'OUTCOME target B EXITED exit={target.p.returncode}; target cwd exists={project.exists()}')
        target.cleanup();target=None
        if initial:initial.cleanup();initial=None
        if child:stop(child,signal.SIGTERM,'unrelated fixture child');child=None
        state(base,'after all cleanup',project);probe(base)
        if row=='F11':
            log('AFTER TARGET AND ALL CLEANUP: restore scratch L mode 0644 solely to hash unchanged bytes')
            (base/'userdata/server-owner.sqlite').chmod(0o644)
            state(base,'post-permission-restore hash verification',project)
        log('ROW COMPLETE '+row)
    finally:
        if target:target.cleanup(signal.SIGKILL)
        if initial:initial.cleanup(signal.SIGKILL)
        if child:stop(child,signal.SIGKILL,'unrelated fixture child emergency cleanup')

def namespace(checkout,phase,run,port):
    base=run/'base';target=None;child=None
    try:
        log('NAMESPACE '+phase+f' init pid={os.getpid()} euid={os.geteuid()} mapped host uid=1000 pidns={os.readlink("/proc/self/ns/pid")}')
        log('uid_map='+pathlib.Path('/proc/self/uid_map').read_text().strip())
        if phase=='A':
            # First child of namespace init is the real owner (PID 2).
            target=Target(checkout,base,run/'owner-project',port,run,'A')
            if target.p.pid!=2:raise RuntimeError('first namespace child not PID 2')
            if target.await_result()!='started':raise RuntimeError('namespace owner failed')
            state(base,'A live immediately before deliberate crash',run/'owner-project')
            probe(base)
            target.cleanup(signal.SIGKILL);target=None
            log('REDEPLOY: original real server A explicitly SIGKILLed and reaped before boot B')
            state(base,'A reaped; released lock left at original inode');probe(base)
        else:
            child=subprocess.Popen(['/bin/sleep','600'],env=ENV);ACTIVE.append(child)
            log(f'SPAWN unrelated redeploy child argv=/bin/sleep 600 tracked namespace pid={child.pid}')
            if child.pid!=2:raise RuntimeError('reused child not PID 2')
            live(child,'unrelated child owns old runtime PID 2')
            project=run/'new-project'
            state(base,'B before launch; stale R and preserved free L',project);probe(base)
            old=json.loads((base/'userdata/server-runtime.json').read_text())
            log(f'PID REUSE PROOF: persisted pid={old["pid"]}; tracked unrelated live pid={child.pid}; match={old["pid"]==child.pid}')
            target=Target(checkout,base,project,port,run,'B')
            result=target.await_result()
            state(base,'B result before cleanup',project);probe(base)
            live(child,'unrelated reused-PID child after CLI')
            log('OUTCOME namespace B '+('STARTED; long-lived, no exit claimed at readiness' if result=='started' else f'EXITED exit={target.p.returncode}; cwd exists={project.exists()}'))
            target.cleanup();target=None
            stop(child,signal.SIGTERM,'unrelated namespace child');child=None
            state(base,'after B and child cleanup',project);probe(base)
    finally:
        if target:target.cleanup(signal.SIGKILL)
        if child:stop(child,signal.SIGKILL,'namespace child emergency cleanup')

def redeploy(checkout,run):
    base=run/'base';base.mkdir();port=freeport()
    state(base,'before namespace A',run/'owner-project');probe(base)
    for phase in ('A','B'):
        argv=['unshare','--user','--map-root-user','--pid','--fork','--mount-proc','--kill-child',sys.executable,SCRIPT,'--checkout',checkout,'--run',run,'--phase',phase,'--port',port]
        log('NAMESPACE LAUNCH '+shlex.join(map(str,argv)))
        p=subprocess.Popen(list(map(str,argv)),env=ENV,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True);ACTIVE.append(p)
        log(f'SPAWN unshare {phase}: tracked host pid={p.pid}')
        sel=selectors.DefaultSelector()
        for name,stream in [('stdout',p.stdout),('stderr',p.stderr)]:os.set_blocking(stream.fileno(),False);sel.register(stream,selectors.EVENT_READ,name)
        end=time.monotonic()+180
        try:
            while sel.get_map():
                if time.monotonic()>end:raise RuntimeError('namespace deadline')
                for key,_ in sel.select(.25):
                    b=os.read(key.fileobj.fileno(),65536)
                    if not b:sel.unregister(key.fileobj);continue
                    for line in b.decode(errors='replace').splitlines():log(f'[namespace {phase} {key.data}] '+line)
            rc=p.wait(timeout=10);log(f'WAIT unshare {phase}: host pid={p.pid} exit={rc}')
            if rc!=0:raise RuntimeError('namespace phase failed '+phase)
        finally:
            if p.poll() is None:stop(p,signal.SIGKILL,'unshare '+phase+' emergency cleanup')
        if phase=='A':
            oldstat=(base/'userdata/server-owner.sqlite').stat()
            log(f'CAPTURE released original L identity: dev={oldstat.st_dev} inode={oldstat.st_ino}')
            state(base,'host after namespace A reaped; before B');probe(base)
        else:
            st=(base/'userdata/server-owner.sqlite').stat()
            log('L ORIGINAL INODE PRESERVED across A crash and B attempt='+str((st.st_dev,st.st_ino)==(oldstat.st_dev,oldstat.st_ino)))
    log('ROW COMPLETE F08')

def main():
    global TRANSCRIPT
    ap=argparse.ArgumentParser();ap.add_argument('--checkout',required=True,type=pathlib.Path);ap.add_argument('--row',choices=['F06','F07','F08','F09','F10','F11']);ap.add_argument('--side');ap.add_argument('--run',type=pathlib.Path);ap.add_argument('--phase');ap.add_argument('--port',type=int);a=ap.parse_args()
    if a.phase:
        namespace(a.checkout,a.phase,a.run,a.port);return
    avail=shutil.disk_usage('/').free
    if avail<5*1024**3:raise RuntimeError(f'FRICTION infra: available disk below 5GB: {avail}')
    stamp=datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%S.%fZ')
    run=ROOT/(stamp+'-'+a.row+'-'+a.side);run.mkdir()
    TRANSCRIPT=open(run/(a.row+'-'+a.side+'.txt'),'w')
    header(a.checkout,a.row,run,a.side)
    try:
        if a.row=='F08':redeploy(a.checkout,run)
        else:ordinary(a.checkout,a.row,run)
    except BaseException:
        log('DRIVER ERROR\n'+traceback.format_exc());raise
    finally:
        for p in ACTIVE:
            if p.poll() is None:stop(p,signal.SIGKILL,'driver final cleanup')
        log('ALL tracked children reaped; no driver-owned process intentionally left running')
        TRANSCRIPT.close()
if __name__=='__main__':main()