#!/usr/bin/env python3 """Synthetic, isolated HTTP/PTY evidence for S31/E02/E03/E04. No worktree edits.""" import argparse, datetime, errno, fcntl, hashlib, json, os, pathlib, pty, re, select, shlex, shutil, signal, struct, subprocess, termios, threading, time from fixture_server import Fixture, artifacts ROOT = pathlib.Path('/tmp/lead-32913-r2/evidence-capture-1') REPO = '/home/user/wt/pr-32913' PINS = {'before': 'fa2f4ff2a2bac5ac92e07749f6b83b0003d642a0', 'after': '1af59792b3ca105da634cdf90c392bae9ad1ff91'} NODE = str(ROOT / 'tools/node-v20.11.1-linux-x64/bin/node') PWSH = str(ROOT / 'tools/powershell-7.4.6/pwsh') ENVIRONMENT = 'Ubuntu 24.04.4 LTS Linux x86_64; dash 0.5.12-6ubuntu5; bash 5.2.21 --posix; curl 8.5.0; PowerShell 7.4.6 Core Linux; Node v20.11.1 (launcher execution), v26.12.1 (real repo build); Python 3.12.3; pipe or controlling PTY as recorded per case' ART = artifacts(NODE) ART_HASHES = {name: {'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()} for name, data in ART.items()} def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='microseconds') def stamp(): return datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%S.%fZ') def write(path, text): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text) return path def source(side, name): return subprocess.check_output(['git', '-C', REPO, 'show', PINS[side] + ':scripts/' + name], text=True) def proc_info(pid): try: status = pathlib.Path(f'/proc/{pid}/status').read_text() return {'pid': pid, 'state': re.search(r'^State:\s*(.*)', status, re.M)[1], 'cmdline': pathlib.Path(f'/proc/{pid}/cmdline').read_bytes().replace(b'\0', b' ').decode(errors='replace').strip()} except FileNotFoundError: return None def make_path(run): path = run / 'path'; path.mkdir() for name in ['cat','rm','grep','sed','head','cut','uname','tar','gzip','sha256sum','wc','tr','awk','mktemp','mkdir','mv','ln','readlink','dirname','pwd','bash','dash','sh','env']: real = shutil.which(name) if real: (path / name).symlink_to(real) for name in ['curl', 'sleep']: real = '/usr/bin/' + name text = '#!/usr/bin/python3\nimport datetime,json,os,sys,time\ne={"utc":datetime.datetime.now(datetime.timezone.utc).isoformat(),"mono":time.monotonic(),"event":"exec ' + name + '","pid":os.getpid(),"pgid":os.getpgrp(),"argv":sys.argv[1:]}\nwith open(os.environ["CAPTURE_TRANSPORT"],"a") as f:f.write(json.dumps(e)+"\\n")\nos.execv(' + repr(real) + ',[' + repr(real) + ']+sys.argv[1:])\n' wrapper = write(path / name, text); wrapper.chmod(0o755) (path / 'node').symlink_to(NODE) return str(path) def ps_program(original, case, origin): prefix = original[:original.index('\nif ($interactive) {\n # Block')] report = '\n[Console]::Error.WriteLine("DISPATCH terminal=$terminal interactive=$interactive errorRedirected=$([Console]::IsErrorRedirected)")\n' if case['mech'] == 'R': a = original.index('$channel = if ($env:T3CODE_CHANNEL)') b = original.index('\nif ($version -match', a) caller = original[a:b].replace('https://github.com', origin).replace('https://api.github.com', origin) return prefix + report + '\n' + caller + '\nWrite-Output "RESOLVED_VERSION=$version"\n' return prefix + report + ''' try { Fetch $env:CAPTURE_URL $env:CAPTURE_DEST -progress Write-Output "DEST_LENGTH=$((Get-Item $env:CAPTURE_DEST).Length)" Write-Output "DEST_SHA256=$((Get-FileHash $env:CAPTURE_DEST -Algorithm SHA256).Hash)" Write-Output "CAPTURE_SUCCESS" } catch { [Console]::Error.WriteLine("EXCEPTION_TYPE=$($_.Exception.GetType().FullName)") [Console]::Error.WriteLine("HTTP_STATUS=$([int]$_.Exception.Response.StatusCode)") [Console]::Error.WriteLine("EXCEPTION=$($_.Exception.Message)") throw } finally { [Console]::Error.WriteLine("CAPTURE_CALLER_FINALLY") } ''' def listing(root): if not root.exists(): return ['ABSENT ' + str(root)] lines = ['DIR ' + str(root)] for path in sorted(root.rglob('*')): st = path.lstat() if path.is_symlink(): lines.append(f'SYMLINK {path} -> {os.readlink(path)}') elif path.is_dir(): lines.append(f'DIR {path} mode={st.st_mode & 0o777:o}') else: lines.append(f'FILE {path} bytes={st.st_size} mode={st.st_mode & 0o777:o}') return lines def capture(case, side, pair, builds): run = pair / side; run.mkdir() for folder in ['home', 'tmp', 'bin']: (run / folder).mkdir() start = time.monotonic(); eventlock = threading.RLock(); events = []; output = []; tracked = set() tf = (run / 'transcript.txt').open('w') sf = (run / 'server-request.log').open('w') def emit(kind, data): with eventlock: ev = {'utc': now(), 't': round(time.monotonic() - start, 6), 'kind': kind, 'data': data} events.append(ev) tf.write(f"{ev['utc']} +{ev['t']:.6f} {kind} {json.dumps(data, ensure_ascii=False)}\n"); tf.flush() if kind == 'SERVER': sf.write(f"{ev['utc']} +{ev['t']:.6f} {json.dumps(data)}\n"); sf.flush() fixture = Fixture(ART); fixture.reset(case, emit) env = {'HOME': str(run / 'home'), 'T3CODE_HOME': str(run / 'home/t3'), 'T3CODE_INSTALL_BIN_DIR': str(run / 'bin'), 'T3CODE_RELEASE_BASE_URL': fixture.origin, 'T3CODE_VERSION': '' if case.get('discovery') else '1.2.3', 'T3CODE_CHANNEL': 'stable', 'TMPDIR': str(run / 'tmp'), 'TMP': str(run / 'tmp'), 'TEMP': str(run / 'tmp'), 'TERM': 'xterm-256color' if case.get('pty') else 'dumb', 'NO_COLOR': '1', 'LANG': 'C.UTF-8', 'LC_ALL': 'C.UTF-8', 'CAPTURE_TRANSPORT': str(run / 'transport.jsonl'), 'POWERSHELL_TELEMETRY_OPTOUT': '1', 'DOTNET_CLI_HOME': str(run / 'home'), 'XDG_CACHE_HOME': str(run / 'home/.cache'), 'XDG_CONFIG_HOME': str(run / 'home/.config'), 'XDG_DATA_HOME': str(run / 'home/.local/share'), 'PROCESSOR_ARCHITECTURE': 'AMD64'} env['PATH'] = make_path(run) # This check is executed with the exact environment subsequently passed to the installer. check = subprocess.check_output(['/bin/sh', '-c', 'env | grep T3CODE; printf "HOME=%s\\n" "$HOME"'], env=env, text=True) assert '/home/user/.t3' not in check assert pathlib.Path(env['HOME']).is_relative_to(ROOT) and pathlib.Path(env['T3CODE_HOME']).is_relative_to(ROOT) emit('ENV_CHECK', {'command': 'env | grep T3CODE; printf HOME', 'output': check}) program = None if case['surface'] == 'S': original = source(side, 'install.sh'); program = original transforms = [] if case.get('discovery'): program = program.replace('https://github.com', fixture.origin).replace('https://api.github.com', fixture.origin) transforms.append('Only hardcoded discovery origin remapped to local synthetic HTTP server; full installer otherwise unchanged.') script = write(run / 'install.sh', program) cmd = (['/bin/dash'] if case['shell'] == 'dash' else ['/bin/bash', '--posix']) + ['-s'] elif case['surface'] == 'P': original = source(side, 'install.ps1'); script = write(run / 'install.ps1', ps_program(original, case, fixture.origin)) transforms = ['Exact prologue and Fetch/retry functions; Linux executes real PS7/.NET HTTP helpers, without claiming a Windows installation.', 'R uses exact release-discovery caller with only GitHub origin remapped.' if case['mech'] == 'R' else 'Synthetic caller invokes Fetch -progress with a real HTTP zip; dispatch chooses IWR or HttpClient according to actual terminal state.'] env['CAPTURE_URL'] = fixture.origin + '/v1.2.3/t3-1.2.3-win32-x64.zip' env['CAPTURE_DEST'] = str(run / 'downloaded.zip') cmd = [PWSH, '-NoLogo', '-NoProfile', '-File', str(script)] else: original = source(side, 'npm-launcher/t3.mjs') launcher = pathlib.Path(builds[side]['launcher']) assert launcher.read_text() == original cmd = [NODE, str(launcher), '--version'] transforms = ['Actual repository buildNpmLauncher output from unpacked npm tarball; launcher bytes verified equal to pinned source; no hooks.'] version = subprocess.check_output([NODE, '--version'], env=env, text=True) emit('NODE_VERSION', {'command': shlex.join([NODE, '--version']), 'output': version}) emit('MANIFEST', {'row': case['row'], 'case': case['name'], 'side': side, 'sha': PINS[side], 'environment': ENVIRONMENT, 'command': shlex.join(cmd), 'env': env, 'source_sha256': hashlib.sha256(original.encode()).hexdigest(), 'executed_source_sha256': hashlib.sha256((program if program is not None else script.read_text() if case['surface']=='P' else original).encode()).hexdigest(), 'transforms': transforms, 'case_definition': case, 'fixture_payloads': ART_HASHES}) master = None; pipe_err = None; pid = None; proc = None; writer = None; input_read = None; input_write = None status = None; sig_sent = False; signal_t = None; signal_requests = None; timedout = False readers = []; transport_seen = 0; launch_info = None try: if case.get('pty'): if program is not None: input_read, input_write = os.pipe() if case.get('redirect_error'): pipe_err = os.pipe() pid, master = pty.fork() if pid == 0: if input_read is not None: os.close(input_write); os.dup2(input_read, 0); os.close(input_read) if pipe_err is not None: os.close(pipe_err[0]); os.dup2(pipe_err[1], 2); os.close(pipe_err[1]) os.chdir(run) os.write(2, f'CAPTURE_FDS stdin_tty={os.isatty(0)} stderr_tty={os.isatty(2)} pid={os.getpid()} pgid={os.getpgrp()}\n'.encode()) os.execve(cmd[0], cmd, env) fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack('HHHH', 30, 120, 0, 0)) readers.append(master) if input_read is not None: os.close(input_read) def send_program(): try: with os.fdopen(input_write, 'wb', buffering=0) as f: f.write(program.encode()) except BrokenPipeError: pass writer = threading.Thread(target=send_program); writer.start() if pipe_err is not None: os.close(pipe_err[1]); readers.append(pipe_err[0]) else: proc = subprocess.Popen(cmd, cwd=run, env=env, stdin=subprocess.PIPE if program is not None else subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=True) pid = proc.pid; readers.append(proc.stdout.fileno()) if program is not None: def send_program(): try: proc.stdin.write(program.encode()); proc.stdin.close() except BrokenPipeError: pass writer = threading.Thread(target=send_program); writer.start() tracked.add(pid) launch_info = {'pid': pid, 'pty': master is not None, 'foreground_pgid': os.tcgetpgrp(master) if master is not None else None, 'stdin': 'full installer piped into shell -s' if program is not None else 'PTY' if master is not None else 'DEVNULL', 'stderr': 'pipe (noninteractive fetch)' if case.get('redirect_error') else 'PTY' if master is not None else 'pipe'} emit('SPAWN', launch_info) for fd in readers: os.set_blocking(fd, False) done = False while not done: ready, _, _ = select.select(readers, [], [], 0.02) if readers else ([], [], []) for fd in ready: try: chunk = os.read(fd, 65536) except OSError as e: if e.errno == errno.EIO: chunk = b'' else: raise if not chunk: readers.remove(fd); continue output.append(chunk); emit('OUTPUT', chunk.decode(errors='replace')) if master is not None and b'\x1b[6n' in chunk: os.write(master, b'\x1b[1;1R' * chunk.count(b'\x1b[6n')); emit('TERMINAL_REPLY', 'cursor row1 column1') transports = [] if (run / 'transport.jsonl').exists(): for line in (run / 'transport.jsonl').read_text().splitlines(): try: transports.append(json.loads(line)) except ValueError: pass for entry in transports[transport_seen:]: emit('TRANSPORT', entry); tracked.add(entry['pid']) transport_seen = len(transports) if case.get('interrupt') and side == 'after' or case.get('interrupt') == 'active': if not sig_sent: if case['interrupt'] == 'backoff': sleeps = [e for e in transports if e['event'] == 'exec sleep' and e['argv'] == ['1']] sleep_state = proc_info(sleeps[-1]['pid']) if sleeps else None condition = sleep_state and sleep_state['state'].startswith('S') and '/usr/bin/sleep 1' in sleep_state['cmdline'] else: sleep_state = None with eventlock: condition = any(e['kind'] == 'SERVER' and e['data'].get('event') == 'body_chunk' and e['data'].get('endpoint') == 'archive' and e['data'].get('offset', 0) > 0 for e in events) condition = condition and 'downloading T3 Code' in b''.join(output).decode(errors='replace') if condition: fg = os.tcgetpgrp(master) assert fg == pid, (fg, pid) emit('DIRECTORY_BEFORE_SIGNAL', listing(run / 'home/t3/runtime/versions')) with fixture.lock: signal_requests = len(fixture.requests) signal_t = time.monotonic() emit('ACTION', {'action': 'Ctrl-C byte to controlling PTY; kernel sends SIGINT to foreground process group', 'foreground_pgid': fg, 'sleep_process': sleep_state, 'request_count_at_signal': signal_requests}) os.write(master, b'\x03'); sig_sent = True if time.monotonic() - start > 45: emit('TIMEOUT', {'action': 'SIGTERM only to tracked spawned process group', 'pid': pid}) os.killpg(pid, signal.SIGTERM); timedout = True if master is not None: _, status = os.waitpid(pid, 0) else: status = proc.wait(timeout=3) done = True elif master is not None: got, child_status = os.waitpid(pid, os.WNOHANG) if got: status = child_status; done = True elif proc.poll() is not None: status = proc.returncode; done = True for fd in list(readers): while True: try: chunk = os.read(fd, 65536) except (OSError, BlockingIOError): break if not chunk: break output.append(chunk); emit('OUTPUT', chunk.decode(errors='replace')) if writer is not None: writer.join(timeout=1) exitcode = os.waitstatus_to_exitcode(status) if master is not None else status if exitcode < 0: exitcode = 128 - exitcode emit('EXIT', {'code': exitcode, 'native_wait_status': status, 'interrupt_sent': sig_sent, 'timeout': timedout}) # Keep the release server listening past the 1s retry boundary after S31's Ctrl-C. # This is an evidence observation window, not a repository timing test. if case['row'] == 'S31': observation_start = time.monotonic() threading.Event().wait(1.5) with fixture.lock: after_count = len(fixture.requests) emit('POST_SIGNAL_OBSERVATION', {'duration_seconds': round(time.monotonic() - observation_start, 6), 'requests_at_signal': signal_requests, 'requests_now': after_count, 'requests_after_signal': after_count - signal_requests if signal_requests is not None else None, 'note': 'Before pin exits on first 504 before any backoff; no signal could be sent.' if not sig_sent else 'Server remained available throughout the observation window.'}) with fixture.settled: fixture.settled.wait_for(lambda: fixture.active == 0, timeout=1.5) emit('SERVER_BEFORE_CLEANUP', {'active_handlers': fixture.active}) # Capture process state and files before shutting down the fixture. remaining = [entry for p in sorted(tracked) if (entry := proc_info(p)) is not None] emit('PROCESS_STATE', {'tracked_pids': sorted(tracked), 'remaining_tracked_processes': remaining, 'remaining_curl': [p for p in remaining if '/usr/bin/curl' in p['cmdline']]}) dirs = listing(run / 'home/t3/runtime/versions') temps = listing(run / 'tmp') (run / 'directory-listing.txt').write_text('\n'.join(dirs + temps) + '\n') emit('DIRECTORY_AFTER_EXIT', dirs); emit('TMP_AFTER_EXIT', temps) state = {'staging': [str(p.relative_to(run)) for p in (run / 'home/t3').rglob('*') if p.is_dir() and '.staging-' in p.name], 'markers': [str(p.relative_to(run)) for p in (run / 'home/t3').rglob('.install-complete')], 'remaining_processes': remaining} with fixture.lock: requests = list(fixture.requests) result = {'row': case['row'], 'case': case, 'side': side, 'sha': PINS[side], 'exit_code': exitcode, 'interrupt_sent': sig_sent, 'timeout': timedout, 'state': state, 'requests': requests, 'request_count_at_signal': signal_requests, 'requests_after_signal': len(requests)-signal_requests if signal_requests is not None else None} if case['surface'] == 'S' and (run / 'bin/t3').exists(): verify = subprocess.run([str(run / 'bin/t3'), '--version'], env=env, capture_output=True, text=True) result['executable_check'] = {'exit': verify.returncode, 'stdout': verify.stdout, 'stderr': verify.stderr} emit('EXECUTABLE_CHECK', result['executable_check']) emit('RESULT', result) write(run / 'output.txt', b''.join(output).decode(errors='replace')) write(run / 'result.json', json.dumps(result, indent=2)) return result finally: if pid is not None: for p in sorted(tracked): info = proc_info(p) if info and not info['state'].startswith('Z'): emit('CLEANUP_TRACKED_PROCESS', info) try: os.kill(p, signal.SIGTERM) except ProcessLookupError: pass fixture.close() emit('FIXTURE_CLOSED', {'port': fixture.server.server_port, 'thread_alive': fixture.thread.is_alive()}) write(run / 'events.json', json.dumps(events, indent=2)) tf.close(); sf.close() if master is not None: os.close(master) if pipe_err is not None: os.close(pipe_err[0]) if proc is not None: proc.stdout.close() def cases(): result = [] for shell in ['dash', 'bash-posix']: for target in ['latest', 'sums']: result.append({'row': 'S31', 'name': f'{target}-{shell}', 'surface': 'S', 'shell': shell, 'target': target, 'seq': [504, 200], 'discovery': target=='latest', 'pty': True, 'redirect_error': True, 'interrupt': 'backoff'}) for interactive in [False, True]: result.append({'row': 'E02', 'name': f'shell-{shell}-' + ('interactive-download' if interactive else 'piped-fetch'), 'surface': 'S', 'shell': shell, 'target': 'archive', 'seq': [502, 'slow' if interactive else 200], 'pty': interactive}) for mech in ['R', 'W', 'H']: result.append({'row': 'E02', 'name': f'pwsh7-{mech}', 'surface': 'P', 'mech': mech, 'target': 'latest' if mech=='R' else 'archive', 'seq': [502, 200], 'discovery': mech=='R', 'pty': mech=='H'}) for target in ['sums', 'archive']: result.append({'row': 'E03', 'name': f'node20-{target}-504', 'surface': 'L', 'target': target, 'seq': [504, 200], 'pty': False}) result.append({'row': 'E04', 'name': 'node20-ctrlc-active-archive', 'surface': 'L', 'target': 'archive', 'seq': ['hold'], 'pty': True, 'interrupt': 'active'}) return result def main(): parser = argparse.ArgumentParser(); parser.add_argument('--rows', default='S31,E02,E03,E04'); args = parser.parse_args() selected = [c for c in cases() if c['row'] in args.rows.split(',')] session = ROOT / ('runs-' + stamp()); session.mkdir() write(session / 'artifact-fingerprints.json', json.dumps(ART_HASHES, indent=2)) write(session / 'environment.txt', ENVIRONMENT + '\n' + subprocess.check_output(['uname','-a'], text=True)) builds = json.loads((pathlib.Path((ROOT / 'latest-build.txt').read_text()) / 'provenance.json').read_text()) results = [] for case in selected: pair = session / (case['row'] + '-' + case['name']); pair.mkdir() for side in PINS: r = capture(case, side, pair, builds); results.append(r) print(f"{case['row']} {case['name']} {side} exit={r['exit_code']} requests=" + ','.join(f"{x['endpoint']}:{x['status']}" for x in r['requests']) + f" staging={r['state']['staging']} signal={r['interrupt_sent']}", flush=True) write(session / 'results.json', json.dumps(results, indent=2)) write(ROOT / 'latest-runs.txt', str(session)) print('SESSION=' + str(session), flush=True) if __name__ == '__main__': main()