rapp-playwright/tools/verify_managed_recovery.py

#!/usr/bin/env python3
"""Approved operator-only crash check for the fixed railiance01 Playwright pilot.

Requires --execute-crash; no identity acceptance or cross-version rollback is implied.
"""
import argparse
from datetime import datetime, timezone
import json
from pathlib import Path
import shlex
import subprocess
import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'client'))
from rapp_playwright import Client, ClientError, collect, save
from verify_crash_recovery import poll, require


def remote(*args):
    r = subprocess.run(['ssh', 'railiance01', shlex.join(list(args))], capture_output=True, text=True, timeout=30)
    require(r.returncode == 0, 'remote check failed; raw output suppressed')
    return json.loads(r.stdout)


def snapshot(expected):
    nodes = remote('kubectl', 'get', 'nodes', '-o', 'json')['items']
    require(any(a['address'] == '92.205.62.239' for n in nodes for a in n['status']['addresses']), 'wrong node')
    pods = remote('kubectl', '-n', 'playwright', 'get', 'pods', '-l', 'app.kubernetes.io/name=playwright', '-o', 'json')['items']
    require(len(pods) == 1, 'expected exactly one pilot pod')
    p = pods[0]
    c = next(c for c in p['status']['containerStatuses'] if c['name'] == 'service')
    require(c['ready'] and c['imageID'] == expected, 'pilot not ready on expected image')
    pvc = remote('kubectl', '-n', 'playwright', 'get', 'pvc', 'playwright-artifacts', '-o', 'json')
    require(pvc['status']['phase'] == 'Bound', 'PVC not bound')
    return {'pod': p['metadata']['name'], 'pod_uid': p['metadata']['uid'], 'restart_count': c['restartCount'],
            'container_id': c['containerID'], 'image_id': c['imageID'], 'pvc_uid': pvc['metadata']['uid'],
            'last_exit_code': c.get('lastState', {}).get('terminated', {}).get('exitCode')}


def verify(args):
    require(args.execute_crash, 'explicit --execute-crash is required after owner/founder approval')
    args.out.mkdir(mode=0o700, parents=True, exist_ok=False)
    report = {'started_at': datetime.now(timezone.utc).isoformat(), 'status': 'failed',
              'authority': 'ADMINISTER @ realm:kubernetes/railiance01', 'activation': 'APPROVED',
              'approval_basis': args.approval_reference, 'identity_acceptance': False,
              'cross_version_rollback': False, 'checks': []}
    client = Client(args.url)
    runs = []
    try:
        before = snapshot(args.image)
        report['before'] = before
        ready = client.request('/v1/preflight')
        require(ready['available'] and ready['release'] == args.image and ready['execution_location'] == 'railiance01/playwright', 'provider mismatch')
        require(ready['active'] == 0 and ready['queued'] == 0, 'pilot busy: do not interrupt unrelated jobs')
        save(args.out / 'preflight.json', ready)
        spec = client.request('/v1/trial')

        def submit(specification, label):
            folder = args.out / label
            folder.mkdir(mode=0o700)
            run = client.request('/v1/runs', 'POST', specification)
            runs.append(run)
            save(folder / 'run.json', run)
            return run, folder

        def finish(run):
            def check():
                r = client.request('/v1/runs/' + run['id'], token=run['token'])
                return r if r['status'] in ('passed', 'failed', 'cancelled', 'interrupted') else None
            return poll(check, 'job completion')

        completed, folder = submit(spec, 'completed-before')
        require(finish(completed)['status'] == 'passed', 'baseline trial failed')
        original = collect(client, completed, folder)
        require(len(original['downloads']) == 4, 'four baseline screenshots required')
        report['checks'].append('baseline_trial')
        slow = {**spec, 'viewports': [{'name': 'desktop', 'width': 1280, 'height': 800}],
                'steps': [{'action': 'screenshot', 'name': 'uncommitted'},
                          {'action': 'visible', 'selector': '#never-present-crash-test'}]}
        active, active_dir = submit(slow, 'active')
        queued, queued_dir = submit(spec, 'queued')
        # Run the guard and signal together on the pod to avoid a multi-command race.
        ids = json.dumps([active['id'], queued['id']])
        code = r'''const fs=require('fs');
(async()=>{
const ids=IDS;
const a=JSON.parse(fs.readFileSync('/data/'+ids[0]+'/receipt.json'));
const q=JSON.parse(fs.readFileSync('/data/'+ids[1]+'/receipt.json'));
if(a.status!=='running'||q.status!=='queued')throw Error('test jobs not active/queued');
const partial='/data/'+ids[0]+'/desktop-0-uncommitted.png';
if(!fs.existsSync(partial)){console.log(JSON.stringify({waiting:true}));return;}
const pf=await (await fetch('http://127.0.0.1:8080/v1/preflight',{headers:{'X-Rapp-Protocol':'1'}})).json();
if(pf.active!==1||pf.queued!==1||pf.release!==IMAGE)throw Error('unexpected concurrent work or release');
const servers=fs.readdirSync('/proc').filter(x=>/^[0-9]+$/.test(x)).filter(x=>{
try{const c=fs.readFileSync('/proc/'+x+'/cmdline','utf8').split(String.fromCharCode(0));return c[0]==='node'&&c[1]==='src/server.mjs';}catch{return false;}});
if(servers.length!==1)throw Error('expected one service supervisor');
const result={signal:'SIGKILL',pid:Number(servers[0]),active_status:a.status,queued_status:q.status,partial_png_bytes:fs.statSync(partial).size};
fs.writeSync(1,JSON.stringify(result));process.kill(Number(servers[0]),'SIGKILL');
})().catch(()=>{process.exitCode=2;});'''.replace('IDS', ids).replace('IMAGE', json.dumps(args.image))

        def crash_when_ready():
            r = subprocess.run(['ssh', 'railiance01', shlex.join(['kubectl', '-n', 'playwright', 'exec', before['pod'], '--', 'node', '-e', code])],
                               capture_output=True, text=True, timeout=30)
            # Killing the supervisor also terminates exec; consume only our JSON receipt.
            try:
                result = json.loads(r.stdout)
            except ValueError:
                raise ClientError('crash guard failed; raw output suppressed') from None
            if result.get('waiting'):
                require(r.returncode == 0, 'crash guard failed while waiting')
                return None
            require(result.get('signal') == 'SIGKILL', 'crash signal receipt absent')
            return result

        report['crash'] = poll(crash_when_ready, 'real in-flight screenshot and guarded crash', timeout=30)
        report['checks'].append('guarded_sigkill_of_active_service')
        # Existing port-forward may end on container restart; the caller restores it.
        def recovered():
            try:
                pf = client.request('/v1/preflight')
                return pf if pf.get('available') and pf.get('release') == args.image else None
            except ClientError:
                return None
        save(args.out / 'after-preflight.json', poll(recovered, 'restore consumer tunnel after container restart', timeout=180))
        def ready_snapshot():
            try:
                return snapshot(args.image)
            except ClientError:
                return None
        after = poll(ready_snapshot, 'ready container after recovery', timeout=60)
        report['after'] = after
        require(after['last_exit_code'] == 137, 'expected SIGKILL termination code')
        require(after['pod_uid'] == before['pod_uid'] and after['pvc_uid'] == before['pvc_uid'], 'pod or storage changed')
        require(after['restart_count'] > before['restart_count'] and after['container_id'] != before['container_id'], 'container restart not observed')
        for run, folder in [(active, active_dir), (queued, queued_dir)]:
            r = collect(client, run, folder)
            require(r['status'] == 'interrupted' and r['error'] == 'service_restarted' and r['artifacts'] == [], 'interruption receipt mismatch')
        check = "const fs=require('fs');console.log(JSON.stringify(" + ids + ".map(id=>fs.readdirSync('/data/'+id))))"
        files = remote('kubectl', '-n', 'playwright', 'exec', before['pod'], '--', 'node', '-e', check)
        require(files == [['receipt.json'], ['receipt.json']], 'orphan artifacts remain')
        report['checks'].append('interrupted_jobs_and_orphan_cleanup')
        folder = args.out / 'completed-after'
        folder.mkdir(mode=0o700)
        restored = collect(client, completed, folder)
        require(restored['status'] == 'passed' and restored['downloads'] == original['downloads'], 'completed artifacts changed')
        report['completed_downloads'] = restored['downloads']
        report['checks'].append('completed_artifact_hashes_preserved')
        fresh, folder = submit(spec, 'new-trial')
        require(finish(fresh)['status'] == 'passed', 'new trial failed')
        r = collect(client, fresh, folder)
        require(r['completed_steps'] == 14 and len(r['downloads']) == 4, 'new trial incomplete')
        report['checks'].append('fresh_trial_after_container_restart')
        report['status'] = 'passed'
    except (ClientError, OSError, ValueError, KeyError, subprocess.TimeoutExpired) as exc:
        report['error'] = str(exc)
    except KeyboardInterrupt:
        report['error'] = 'interrupted_by_caller'
    finally:
        for run in runs:
            try:
                client.request('/v1/runs/' + run['id'], 'DELETE', token=run['token'])
            except ClientError:
                pass
        report['finished_at'] = datetime.now(timezone.utc).isoformat()
        save(args.out / 'verification.json', report)
    print(json.dumps(report, indent=2))
    return 0 if report['status'] == 'passed' else 3


if __name__ == '__main__':
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--url', required=True)
    p.add_argument('--image', required=True)
    p.add_argument('--out', type=Path, required=True)
    p.add_argument('--approval-reference', required=True)
    p.add_argument('--execute-crash', action='store_true')
    try:
        sys.exit(verify(p.parse_args()))
    except (ClientError, OSError, ValueError) as exc:
        print(json.dumps({'error': str(exc)}), file=sys.stderr)
        sys.exit(3)