#!/usr/bin/env python3 import argparse import hashlib import json from pathlib import Path import shlex import signal import sys import tempfile import time import uuid from fs import session def screen_text(client): screen = client.screen() cols = screen['cols'] return '\n'.join(''.join(cell[0] for cell in screen['cells'][at:at + cols]) for at in range(0, len(screen['cells']), cols)) def expired(signum, frame): raise TimeoutError('session deadline exceeded') def wait_ready(client, ready): previous = None while True: text = screen_text(client) if ready in text and text == previous: return text previous = text time.sleep(.05) def run(args, command, status): temporary = tempfile.TemporaryDirectory(prefix='pardes-agent-session-') try: root = Path(temporary.name) config = root / 'config' / 'pardes' config.mkdir(parents=True) (config / 'init').write_text('Shell /bin/sh\n') script = root / 'command.sh' script.write_text('exec ' + shlex.join(command) + '\n') old_handler = signal.signal(signal.SIGALRM, expired) signal.setitimer(signal.ITIMER_REAL, args.timeout) try: context = session(str(args.binary.resolve()), root, 'agent-session', tty=True, socket_name='agent-session', launch=['--tty', '--9p=agent-session'], inherited={'PARDES_TEST_GRID': '1'} if args.gui_grid else {}) client, _ = context.__enter__() try: try: while not client.read('/self/pane/1/body').strip(): time.sleep(.01) started = time.monotonic() client.write('/self/pane/1/pty/data', shlex.join(['/bin/sh', str(script)]).encode() + b'\r') status['phase'] = 'readiness' wait_ready(client, args.ready) if args.prepare_key is not None: status['phase'] = 'preparation' client.write('/self/pane/1/pty/data', args.prepare_key) wait_ready(client, args.ready) ready_ms = (time.monotonic() - started) * 1000 status.update(phase='history', required_body_rows=args.min_rows) while True: started = time.monotonic() body = client.read('/self/pane/1/body') body_read_ms = (time.monotonic() - started) * 1000 rows = body.count(b'\n') + int(bool(body) and not body.endswith(b'\n')) status.update(body_bytes=len(body), body_rows=rows) if rows >= args.min_rows: break time.sleep(.05) status['phase'] = 'settling' before = wait_ready(client, args.ready) status['expected_visible_text_sha256'] = hashlib.sha256(before.encode()).hexdigest() probe = 'p' + uuid.uuid4().hex[:12] if probe in before: raise AssertionError('probe is already visible') status['phase'] = 'probe-visible' started = time.monotonic() client.write('/self/pane/1/pty/data', probe.encode()) while probe not in screen_text(client).replace('\n', ''): time.sleep(.01) write_ms = (time.monotonic() - started) * 1000 status['phase'] = 'probe-cleared' started = time.monotonic() client.write('/self/pane/1/pty/data', b'\x15') while True: observed = screen_text(client) status['observed_visible_text_sha256'] = hashlib.sha256(observed.encode()).hexdigest() if observed == before: break time.sleep(.01) status['metrics'] = { 'ready_ms': ready_ms, 'body_read_ms': body_read_ms, 'ninep_write_to_observation_ms': write_ms, 'ninep_clear_to_observation_ms': (time.monotonic() - started) * 1000, 'visible_text_sha256': hashlib.sha256(before.encode()).hexdigest(), 'body_bytes': len(body), 'body_rows': rows, 'body_sha256': hashlib.sha256(body).hexdigest(), } finally: previous_phase = status['phase'] status['phase'] = 'interrupt-cleanup' signal.setitimer(signal.ITIMER_REAL, 0) client.socket.settimeout(.25) for _ in range(2): try: client.write('/self/pane/1/pty/data', b'\x03') except (OSError, EOFError): pass time.sleep(.05) status['phase'] = previous_phase finally: previous_phase = status['phase'] status['phase'] = 'session-cleanup' context.__exit__(*sys.exc_info()) status['phase'] = previous_phase finally: signal.setitimer(signal.ITIMER_REAL, 0) signal.signal(signal.SIGALRM, old_handler) finally: previous_phase = status['phase'] status['phase'] = 'files-cleanup' temporary.cleanup() status['phase'] = previous_phase return status.pop('metrics') def main(argv): parser = argparse.ArgumentParser(description='Check an unsubmitted input probe through an owned Pardes 9P session.') parser.add_argument('binary', type=Path) parser.add_argument('--ready', required=True, help='visible application readiness text') parser.add_argument('--timeout', type=float, default=60, help='interaction deadline in seconds, at most 600') parser.add_argument('--min-rows', type=int, default=0) parser.add_argument('--prepare-key', help='one hex control byte before history/probe checks; never CR or LF') parser.add_argument('--gui-grid', action='store_true', help='use the GUI build headless grid host') if '--' not in argv: parser.error('supply the interactive command after --') split = argv.index('--') args = parser.parse_args(argv[:split]) command = argv[split + 1:] if not command or not command[0] or not args.ready.strip(): parser.error('a command and nonempty --ready text are required') if not 0 < args.timeout <= 600 or args.min_rows < 0: parser.error('--timeout must be in (0, 600] and --min-rows must be nonnegative') if args.prepare_key is not None: try: args.prepare_key = bytes.fromhex(args.prepare_key) except ValueError: parser.error('--prepare-key must be one hex control byte') if len(args.prepare_key) != 1 or not 0 < args.prepare_key[0] < 32 or args.prepare_key[0] in (10, 13): parser.error('--prepare-key must be one control byte other than CR or LF') status = {'phase': 'startup'} try: metrics = run(args, command, status) except (OSError, EOFError, AssertionError, ValueError) as error: status['error'] = type(error).__name__ if isinstance(error, OSError): status['errno'] = error.errno print(json.dumps(status), file=sys.stderr) return 1 print(json.dumps(metrics)) return 0 if __name__ == '__main__': sys.exit(main(sys.argv[1:]))