#!/usr/bin/env python3 """Event-gated synthetic production-runtime recordings. No T3 database is used.""" import argparse import hashlib import datetime import json import os import pathlib import platform import selectors import signal import subprocess import time SCRATCH = pathlib.Path('/tmp/lead-33628-r1/capture-1') parser = argparse.ArgumentParser() parser.add_argument('worktree') parser.add_argument('side', choices=['before', 'after']) parser.add_argument('row', choices=['A1', 'A2', 'A3', 'B1', 'B2', 'B3', 'C2-C3']) args = parser.parse_args() sha = subprocess.check_output(['git', '-C', args.worktree, 'rev-parse', 'HEAD'], text=True).strip() stamp = datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%S.%fZ') out = SCRATCH / f'{args.row}-{args.side}-{stamp}' out.mkdir() script_hashes = {name: hashlib.sha256((SCRATCH / name).read_bytes()).hexdigest() for name in ['driver.ts', 'cloudflared-stub.sh', 'cloudflared-stub.mjs', 'runner.py', 'run.sh']} node_version = subprocess.check_output(['node', '--version'], cwd=args.worktree + '/apps/server', text=True).strip() slow = {'A1': 1000, 'A2': 1000, 'A3': 0, 'B1': 0, 'B2': 1000, 'B3': 1600, 'C2-C3': 0}[args.row] env = dict(os.environ, WT=args.worktree, ROW=args.row, SLOW_MS=str(slow), NO_COLOR='1', STUB=str(SCRATCH / 'cloudflared-stub.sh'), STUB_JS=str(SCRATCH / 'cloudflared-stub.mjs'), BASE_DIR=str(out / 't3home'), STUB_LOG=str(out / 'stub.log'), STUB_STATE=str(out / 'stub-state')) runner_log = open(out / 'runner.log', 'w') driver_log = open(out / 'driver.log', 'wb') events = [] raw_buffer = b'' stop_ms = None failed = None selector = selectors.DefaultSelector() proc = None def log(event, **data): item = dict(timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='milliseconds'), source='runner', event=event, **data) runner_log.write(json.dumps(item) + '\n') runner_log.flush() print(json.dumps(item), flush=True) return item def decode(line): try: item = json.loads(line) if isinstance(item, dict): if item.get('event') == 'file-line': try: item['record'] = json.loads(item['line']) except json.JSONDecodeError: pass events.append(item) except (json.JSONDecodeError, UnicodeDecodeError): pass def pump(timeout=1): global raw_buffer for key, _ in selector.select(timeout): chunk = os.read(key.fd, 65536) if not chunk: selector.unregister(key.fileobj) continue driver_log.write(chunk) driver_log.flush() raw_buffer += chunk while b'\n' in raw_buffer: line, raw_buffer = raw_buffer.split(b'\n', 1) decode(line) def wait(predicate, description, timeout=25, since=0): deadline = time.monotonic() + timeout while True: for event in events[since:]: if predicate(event): log('gate-satisfied', gate=description, observed=event) return event if time.monotonic() >= deadline: raise RuntimeError('Timed out awaiting ' + description) if proc.poll() is not None and not selector.get_map(): raise RuntimeError('Driver exited before ' + description) pump(min(1, max(0, deadline - time.monotonic()))) def driver(event, **fields): return lambda e: e.get('source') == 'driver' and e.get('event') == event and all(e.get(k) == v for k, v in fields.items()) def stub(event, attempt): return lambda e: e.get('source') == 'stub' and e.get('record', {}).get('event') == event and e['record'].get('attempt') == attempt def relay(message): return lambda e: e.get('source') == 'relay' and message in str(e.get('record', {}).get('message', '')) def signal_child(attempt): e = wait(stub('started', attempt), f'stub attempt {attempt} started') pid = e['record']['pid'] log('send-child-only-signal', pid=pid, signal='SIGUSR1', purpose='controlled nonzero exit 1, no server signal') os.kill(pid, signal.SIGUSR1) return pid def send_parent(sig): log('send-parent-only-signal', pid=proc.pid, signal=signal.Signals(sig).name) os.kill(proc.pid, sig) wait(driver('host-signal-received', signal=signal.Signals(sig).name), 'host receipt of ' + signal.Signals(sig).name) def shutdown_group(sig): global stop_ms start = time.monotonic_ns() log('send-group-signal', pid=proc.pid, pgid=proc.pid, signal=signal.Signals(sig).name) os.killpg(proc.pid, sig) wait_exit() stop_ms = round((time.monotonic_ns() - start) / 1_000_000) log('backend-stopped', stopMs=stop_ms, returncode=proc.returncode) def wait_exit(timeout=25): deadline = time.monotonic() + timeout while selector.get_map(): if time.monotonic() >= deadline: raise RuntimeError('Driver did not exit') pump(min(1, max(0, deadline - time.monotonic()))) proc.wait(timeout=max(0.1, deadline - time.monotonic())) def command(value): log('send-driver-command', command=value) proc.stdin.write(value + '\n') proc.stdin.flush() def group_members(): members = [] for entry in pathlib.Path('/proc').iterdir(): if not entry.name.isdigit(): continue try: stat = (entry / 'stat').read_text().rsplit(')', 1)[1].split() if int(stat[2]) == proc.pid: members.append({'pid': int(entry.name), 'state': stat[0]}) except (OSError, ValueError, IndexError): pass return members try: log('environment', row=args.row, side=args.side, sha=sha, os=platform.platform(), node=node_version, harness='real CloudManagedEndpointRuntime + real Node spawner with passive exit observer + stub cloudflared; not a real Cloudflare tunnel', attribution='AI-assisted; verified with the tests and tools listed above.', slowMs=slow) proc = subprocess.Popen(['node', str(SCRATCH / 'driver.ts')], cwd=args.worktree + '/apps/server', env=env, start_new_session=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) log('spawned-driver', pid=proc.pid, pgid=proc.pid) selector.register(proc.stdout, selectors.EVENT_READ) if args.row.startswith('A'): wait(driver('driver-ready'), 'runtime applied and driver ready') wait(stub('past-registering-window', 1), 'original stub age > 2 seconds (synthetic marker, not tunnel readiness)') shutdown_group(signal.SIGINT if args.row == 'A2' else signal.SIGTERM) elif args.row == 'B1': wait(driver('driver-ready'), 'runtime applied; no server signal') signal_child(1) wait(driver('recovery-request', count=1), 'ordinary crash recovery request') wait(stub('past-registering-window', 2), 'replacement observed and aged beyond synthetic registering window') log('ordinary-crash-recovered-before-server-signal') shutdown_group(signal.SIGTERM) elif args.row == 'B2': wait(driver('reconcile-lock-held'), 'public restartConnector holds reconcile lock') pid = signal_child(1) wait(driver('child-exit-observed', pid=pid, exitCode=1), 'real exit effect observed while reconcile lock is held') log('ordering-established', ordering='lock held -> child exit observed -> group signal -> gate release -> supervisor decision') shutdown_group(signal.SIGTERM) elif args.row == 'B3': wait(driver('driver-ready'), 'runtime applied; no server signal') signal_child(1) wait(driver('recovery-request', count=1), 'first crash recovery establishes next backoff') signal_child(2) delay = wait(relay('crash-looping; delaying restart'), 'recorded positive crash-loop backoff') log('signal-during-recorded-backoff', delayRecord=delay['record'], predecessorMs=slow) shutdown_group(signal.SIGTERM) else: for phase, attempt in [('C2-no-signal', 1), ('C3-signalled', 2), ('C3-fresh', 3)]: wait(driver('phase-ready', phase=phase), phase + ' runtime built') wait(stub('past-registering-window', attempt), phase + ' original beyond synthetic registering window') if phase == 'C3-signalled': send_parent(signal.SIGINT) send_parent(signal.SIGTERM) log('host-survived-both-signals', explanation='Effect.runPromise lifecycle host intentionally keeps running; not NodeRuntime signal shutdown') if phase == 'C3-fresh': signal_child(attempt) wait(driver('recovery-request', phase=phase), 'fresh runtime ordinary crash recovery after previous layer was signalled and closed') wait(stub('past-registering-window', attempt + 1), 'fresh runtime replacement observed') command('close:' + phase) wait(driver('phase-closed', phase=phase), phase + ' explicitly closed') wait(driver('lifecycle-complete'), 'all lifecycle phases complete') wait_exit() log('lifecycle-host-stopped', returncode=proc.returncode) members = group_members() log('tracked-process-group-cleanup', members=members) if members: raise RuntimeError('Tracked group still contains processes') except BaseException as exc: failed = str(exc) log('capture-error', reason=failed) if proc is not None: try: os.killpg(proc.pid, signal.SIGTERM) log('cleanup-TERM-tracked-group', pgid=proc.pid) wait_exit(8) except BaseException: try: os.killpg(proc.pid, signal.SIGKILL) log('cleanup-KILL-tracked-group', pgid=proc.pid) except ProcessLookupError: pass proc.wait() finally: runner_log.close() driver_log.close() selector.close() relay_path = out / 't3home/userdata/logs/relay-client.ndjson' if relay_path.exists(): (out / 'relay-client.ndjson').write_bytes(relay_path.read_bytes()) summary = dict(row=args.row, side=args.side, sha=sha, stopMs=stop_ms, failure=failed, output=str(out), scriptSha256=script_hashes, recoveryRequests=[e for e in events if driver('recovery-request')(e)], listeners=[e for e in events if driver('listeners')(e)], scopeStops=[e for e in events if driver('explicit-scope-close-end')(e)]) (out / 'summary.json').write_text(json.dumps(summary, indent=2) + '\n') relay_records = [json.loads(line) for line in (out / 'relay-client.ndjson').read_text().splitlines()] if (out / 'relay-client.ndjson').exists() else [] summary['relayCounts'] = {message: sum(isinstance(item.get('message'), list) and item['message'][0] == message for item in relay_records) for message in ['Relay client process started; waiting for tunnel connection', 'Relay client exited; restarting', 'Relay client stopped', 'Relay client is crash-looping; delaying restart']} summary['exitResults'] = [e for e in events if driver('child-exit-observed')(e)] (out / 'summary.json').write_text(json.dumps(summary, indent=2) + '\n') parts = [f'# {args.row} {args.side} runtime recording\n\n', f'Revision: [{sha[:10]}](https://git.cafe/pingdotgg/t3code/commit/{sha})\n\n', f'Environment: {platform.platform()}, Node {node_version}. Production-runtime harness, synthetic config and stub connector. Real CloudManagedEndpointRuntime and Node spawner (passive exit-result observer). Not a real Cloudflare tunnel.\n\n', 'A/B: NodeRuntime.runMain, group signal. C2/C3: Effect.runPromise host intentionally survives parent-only OS signals, explicit runtime scope closure.\n\n', 'Stub registration is an age-based model only: <2000ms stop takes 2300ms, otherwise 15ms. “past-registering-window” is synthetic, never Cloudflare readiness. Runner gates use events, not guessed readiness sleeps. The predecessor/backoff/stub timers are deliberate fixture behavior.\n\n', 'AI-assisted; verified with the tests and tools listed above.\n\n', '## Measured result\n\n```json\n' + json.dumps(summary, indent=2) + '\n```\n\n'] for title, filename in [('Runner milestones', 'runner.log'), ('Driver and forwarded timestamped relay/stub events', 'driver.log'), ('Full raw relay-client log', 'relay-client.ndjson'), ('Full raw stub log', 'stub.log')]: path = out / filename parts += [f'## {title}\n\n```text\n', path.read_text() if path.exists() else '(file absent)\n', '```\n\n'] (out / f'{args.row}-{args.side}-transcript.md').write_text(''.join(parts)) print('CAPTURE_DIR=' + str(out), flush=True) if failed: raise SystemExit(1)