XFESystem

"""Verify a source-bound independent kernel in a private QEMU/UEFI guest."""
import argparse
import hashlib
import json
import re
from pathlib import Path
import shutil
import socket
import subprocess
import sys
import time
from PIL import Image
from BootAnimationEvidence import BootAnimationCapture

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / 'Tools' / 'CSharpBoot'))
sys.path.insert(0, str(ROOT / 'Tools'))
from ReadImage import inspect_image
from TestBrowserPipeline import _monaco_frame_stats, _monaco_fourth_line_delta
from IndependentDocumentEvidence import PRESENTED, run_documents, seed_documents, window_geometry
from IndependentBingEvidence import json_blocks, native_network_profile, network_arguments, run_bing, validate_network_arguments
from GpuDiscoveryEvidence import validate_discovery
from IndependentConsoleEvidence import assert_console_isolation, run_console, terminal_command_output



def digest(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def read_json(path):
    return json.loads(Path(path).read_text(encoding='utf-8-sig'))


def boot_menu_labels_present(serial):
    # Limine's serial UI can position the cursor before each individual
    # character. Strip only CSI control sequences for label recognition;
    # the original byte-for-byte serial evidence remains on disk.
    displayed = re.sub(r'\x1b\[[0-?]*[ -/]*[@-~]', '', serial)
    return 'XFESystem Desktop' in displayed and 'XFESystem Console' in displayed


def run_terminal_commands(proof, text, healthy, screenshot, key, type_text, input_kind='PS2'):
    """Real PS/2 CLI acceptance. No host storage reads or injected shell execution."""
    started = time.monotonic(); beginning = len(text()); selected = []
    window_id, _ = window_geometry(healthy(), 'Terminal')
    result = {'Kind': 'XfeIndependentClassicTerminalWorkflow', 'Passed': False,
              'Input': f'Actual QMP {input_kind} keyboard packets only', 'WindowId': window_id,
              'HostCommandExecution': False, 'HostWorkspaceReads': False, 'Commands': []}
    output_path = proof / 'TerminalWorkflow.json'

    def wait_command(after, command, required_lines=()):
        deadline = time.monotonic() + 30
        last_error = 'No fresh complete rendered terminal snapshot'
        while time.monotonic() < deadline:
            packets = [item for item in json_blocks(healthy(), 'XFE_TERMINAL_SNAPSHOT', after)
                       if item['Value'].get('WindowId') == window_id]
            if packets:
                packet = packets[-1]
                try:
                    lines = terminal_command_output(packet['Value'], command)
                    assert all(required in lines for required in required_lines), 'Missing actual post-command output: ' + repr(required_lines)
                    selected.append(packet)
                    result['Commands'].append({'Command': command, 'Passed': True, 'SerialOffset': packet['SerialOffset'],
                        'Prompt': packet['Value']['Prompt'], 'Version': packet['Value']['Version'], 'Busy': False,
                        'PostCommandOutput': lines})
                    return packet['Value'], lines
                except AssertionError as error:
                    last_error = str(error)
            time.sleep(.1)
        raise AssertionError('Actual terminal command timeout: ' + command + ': ' + last_error)

    def execute(command, required_lines=()):
        after = len(text()); type_text(command); key('ret')
        return wait_command(after, command, required_lines)

    try:
        snapshot, help_lines = execute('help')
        assert any('md Path | mf Path | cat Path | write Path Text' in line for line in help_lines), 'The actual CLI help lacks real file commands'
        assert any('Up/Down: history' in line for line in help_lines), 'The actual CLI help lacks history/caret editing'
        assert 'Run Demo' not in snapshot['Text'], 'The removed demo-only terminal content is still present'
        result['HelpPassed'] = True
        execute('md CliProof')
        execute('mf CliProof/Command.txt')
        write_command = 'write CliProof/Command.txt NativeTerminalReady'
        execute(write_command)
        execute('cat CliProof/Command.txt', ('NativeTerminalReady',))
        result['RamDirectoryCreateWriteReadPassed'] = True

        # Go beyond the most recent command. Executing the recalled command
        # must produce its exact identity after the actual native render.
        after = len(text()); key('up'); key('up'); key('ret')
        wait_command(after, write_command)
        result['HistoryBeyondPreviousPassed'] = True
        after = len(text()); type_text('echo draft'); key('up'); key('down'); key('ret')
        wait_command(after, 'echo draft', ('draft',))
        result['HistoryDraftRestorePassed'] = True

        # Home/right/delete insert a missing scalar; left plus overwrite mode
        # replaces the last scalar. The submitted command and its own output
        # both need to match, rather than searching previous echo history.
        after = len(text()); type_text('echo aXc'); key('home')
        for _ in range(6): key('right')
        key('delete'); type_text('b'); key('right'); key('left'); key('insert'); type_text('d'); key('insert'); key('end'); key('ret')
        wait_command(after, 'echo abd', ('abd',))
        result['CaretDeleteInsertEditingPassed'] = True
        _, history_lines = execute('history')
        assert any(write_command in line for line in history_lines) and any('echo abd' in line for line in history_lines), 'Actual history omitted executed commands'
        result['HistoryListingPassed'] = True
        execute('cat /System/Diagnostics/Gpu.txt', ('Scope: XFESystemGuest', 'CUDA: unavailable',
            'GPU rasterization: unavailable', 'Discovery state: 2; devices: 0',
            'GB202 boot probe state: 1; FSP scratch 0x00000000'))
        result['NativeGpuDiagnosticRamFileReadbackPassed'] = True
        result['GpuDiagnosticScreenshot'] = {'Path': 'Terminal-GpuDiagnostics.png', 'Sha256': screenshot('Terminal-GpuDiagnostics')}
        result['Screenshot'] = {'Path': 'Terminal-ClassicCLI.png', 'Sha256': screenshot('Terminal-ClassicCLI')}
        result['NoDemoOnlyTerminalContent'] = True
        result['Passed'] = True
    except Exception as error:
        result['Failure'] = str(error)
        raise
    finally:
        raw = proof / 'TerminalCommandEvidence.log'
        raw.write_text('\n'.join(item['Raw'] for item in selected), encoding='utf-8')
        result['RawGuestEvidence'] = {'Path': raw.name, 'Sha256': digest(raw), 'SelectedRenderedSnapshots': len(selected), 'StartSerialOffset': beginning}
        result['ElapsedHostSeconds'] = time.monotonic() - started
        output_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
    return result


class Qmp:
    def __init__(self, port):
        self.connection = socket.create_connection(('127.0.0.1', port), timeout=10)
        self.stream = self.connection.makefile('rwb')
        self.number = 0
        greeting = json.loads(self.stream.readline())
        assert 'QMP' in greeting, greeting
        self.call('qmp_capabilities')

    def call(self, command, arguments=None):
        self.number += 1
        item = {'execute': command, 'id': self.number}
        if arguments is not None:
            item['arguments'] = arguments
        self.stream.write(json.dumps(item).encode() + b'\n')
        self.stream.flush()
        while True:
            line = self.stream.readline()
            if not line:
                raise RuntimeError('QEMU closed QMP unexpectedly')
            result = json.loads(line)
            if result.get('id') == self.number:
                if 'error' in result:
                    raise RuntimeError(result['error'])
                return result.get('return')

    def close(self):
        self.stream.close()
        self.connection.close()


def exercise_desktop(qmp, proof, text, browser_required, documents_required=False, bing_required=False, input_kind='PS2'):
    pointer = [640, 400]
    def healthy():
        log = text()
        assert not any(marker in log for marker in ('XFE_NATIVE_FAIL', 'XFE_NATIVE_EXCEPTION', 'XFE_RUNTIME_FATAL',
                                                     'XFE_ZERO_FAILFAST', 'XFE_INDEPENDENT_APPLICATION_FAILED',
                                                     'XFE_BROWSER_PAINT_FAILED', 'Rendering failed:',
                                                     'XFE_EDITOR_BACKGROUND|SessionError|')), log[-4096:]
        return log
    def wait(marker, start=0, seconds=90):
        end = time.monotonic() + seconds
        while time.monotonic() < end:
            log = healthy()
            if marker in log[start:]:
                return
            time.sleep(.1)
        raise AssertionError('Desktop acknowledgement timeout: ' + marker + '\n' + healthy()[-4096:])
    def screenshot(name):
        path = proof / (name + '.ppm')
        qmp.call('screendump', {'filename': str(path)})
        with Image.open(path) as frame:
            frame.convert('RGB').save(proof / (name + '.png'))
        return digest(proof / (name + '.png'))
    def key(chord):
        qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord})
        time.sleep(.15)
    def type_text(value):
        special = {':': 'shift-semicolon', '.': 'dot', '/': 'slash', '-': 'minus', '_': 'shift-minus', ' ': 'spc', '\n': 'ret'}
        for char in value:
            key(special.get(char, 'shift-' + char.lower() if char.isupper() else char))
    def move(x, y):
        while pointer != [x, y]:
            dx, dy = max(-80, min(80, x-pointer[0])), max(-80, min(80, y-pointer[1]))
            qmp.call('input-send-event', {'events': [
                {'type': 'rel', 'data': {'axis': 'x', 'value': dx}},
                {'type': 'rel', 'data': {'axis': 'y', 'value': dy}}]})
            pointer[0] += dx; pointer[1] += dy
            time.sleep(.025)
    def click(x, y):
        move(x, y)
        for down in (True, False):
            qmp.call('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'left'}}]})
            time.sleep(.1)
    def wheel(count):
        for _ in range(count):
            for down in (True, False):
                qmp.call('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'wheel-down'}}]})
                time.sleep(.1)
    if 'XFE_DESKTOP_ACTIVE|Terminal' in healthy():
        start = len(text()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', start)
    else:
        wait('XFE_DESKTOP_ACTIVE|Files')
    files_hash = screenshot('Desktop-Files')
    start = len(text()); click(370, 116)
    wait('XFE_DESKTOP_ACTIVE|Editor', start)
    type_text('hello')
    start = len(text()); key('ctrl-s')
    wait('XFE_DESKTOP_SAVED|X:\\Users\\Default\\Documents\\Document1.txt|5', start)
    wait('XFE_DESKTOP_SAVED_TEXT_BEGIN\nhello\nXFE_DESKTOP_SAVED_TEXT_END', start)
    editor_hash = screenshot('Desktop-Editor-Saved')
    assert files_hash != editor_hash, 'Editor input produced no changed guest framebuffer'
    start = len(text()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', start)
    start = len(text()); click(220, 170); wait('XFE_DESKTOP_ACTIVE|Editor', start)
    start = len(text()); key('ctrl-s')
    wait('XFE_DESKTOP_SAVED|X:\\Users\\Default\\Documents\\Document1.txt|5', start)
    wait('XFE_DESKTOP_SAVED_TEXT_BEGIN\nhello\nXFE_DESKTOP_SAVED_TEXT_END', start)
    start = len(text()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', start)
    start = len(text()); key('ctrl-esc'); wait('XFE_DESKTOP_FLAGS|1|0|', start)
    key('down'); key('down'); key('ret')
    wait('XFE_DESKTOP_ACTIVE|Browser', start)
    wait('XFE_DESKTOP_STATE_END', start)
    bing_proof = None
    if bing_required:
        # The actual menu input above opened the product's unchanged Bing
        # default. Never substitute a host download or offline page here.
        bing_proof = run_bing(proof, text, healthy, wait, screenshot, key, type_text, move,
                              open_browser=False, start_offset=start)
    browser_id, _ = window_geometry(healthy(), 'Browser')
    # Let the genuine default navigation publish before editing its address;
    # its completion writes the page URL back to the same address model.
    wait('XFE_EDITOR_BACKGROUND|Ready|Window|' + str(browser_id) + '|', start, seconds=180)
    # The merged desktop preserves its Bing default URL. Exercise the built-in
    # JavaScript page through the actual address bar instead of assuming it is
    # the newly opened page or changing the product default for the harness.
    start = len(text()); key('ctrl-l'); type_text('about:Start'); key('ret')
    wait('JavaScript ready: 12', start)
    browser_hash = screenshot('Browser-JavaScript')
    assert browser_hash not in (files_hash, editor_hash), 'Browser did not present a distinct guest frame'
    details = {'MouseCreatedEditor': True, 'KeyboardSavedRamFile': True, 'SavedRamFileReopened': True, 'AltF4ClosedEditor': True,
               'StartMenuKeyboardNavigation': True, 'JavaScriptDomResult': 'JavaScript ready: 12',
               'Screenshots': {'Files': files_hash, 'EditorSaved': editor_hash, 'Browser': browser_hash}}
    if bing_required:
        assert bing_proof['Passed'] is True and bing_proof['NativeHttpsBingLoaded'] is True
        details['BingEvidence'] = {'Path': 'BingEvidence.json', 'Sha256': digest(proof / 'BingEvidence.json'),
                                  'Passed': True, 'NativeHttpsBingLoaded': True}
    if browser_required:
        start = len(text()); key('ctrl-l'); type_text('about:Monaco'); key('ret')
        model_pattern = re.compile(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END', re.S)
        # The background protocol publishes model snapshots rather than the
        # page's DOM status label. Require the actual three-line initial model.
        expected_initial = '// XFESystem | XFEstudio\nconst answer = 6 * 7;\nconsole.log(answer);'
        deadline = time.monotonic() + 180
        models = []
        while time.monotonic() < deadline:
            models = model_pattern.findall(healthy()[start:])
            if models and models[-1] == expected_initial: break
            time.sleep(.1)
        assert models and models[-1] == expected_initial, 'The actual initial Monaco model is unavailable or differs'
        details['MonacoInitialModel'] = models[-1]
        # Background page owners publish immutable paint snapshots. The UI
        # must acknowledge actual scanout before glyph evidence can qualify.
        window_id, _ = window_geometry(healthy(), 'Browser')
        deadline = time.monotonic() + 90
        presented = None
        while time.monotonic() < deadline:
            matches = [item for item in PRESENTED.finditer(healthy(), start)
                       if int(item[1]) == window_id and '/Monaco/' in item[3] and int(item[5]) > 0 and int(item[6]) > 0]
            if matches:
                presented = matches[-1]; break
            time.sleep(.1)
        assert presented is not None, 'The actual Monaco snapshot was not presented to scanout'
        details['MonacoPresentation'] = {'WindowId': window_id, 'Address': presented[3],
            'SnapshotSequence': int(presented[4]), 'PresentationSequence': int(presented[5]), 'PresentationTicks': int(presented[6])}
        details['MonacoRasterPaintCompleted'] = True
        windows = re.findall(r'XFE_WINDOW_STATE\|\d+\|Browser\|(-?\d+)\|(-?\d+)\|(\d+)\|(\d+)\|', text())
        assert windows, 'Actual browser window geometry is missing'
        left, top, width, height = map(int, windows[-1])
        viewport = (left + 2, top + 67, width - 4, height - 96)
        move(left + width - 20, top + 50)
        def paint_frame(name, edited=False, before=None):
            end = time.monotonic() + 60
            attempt = 0
            while time.monotonic() < end:
                healthy(); attempt += 1
                screenshot(name + '-' + str(attempt))
                with Image.open(proof / (name + '-' + str(attempt) + '.png')) as pixels:
                    frame = pixels.copy()
                stats = _monaco_frame_stats(frame, viewport, 1)
                if edited:
                    delta = _monaco_fourth_line_delta(before, frame, stats['Lines'][3]['NativeBounds'])
                    stats['FourthLineDelta'] = delta
                    visible = stats['Lines'][3]['HasTextGlyphDistribution'] and delta['NewInkPixels'] >= 35 and delta['NewInkColumns'] >= 24
                else:
                    visible = stats['ReadyPainted']
                if visible:
                    return frame, stats
                time.sleep(.25)
            raise AssertionError('Actual Monaco glyph/syntax pixels did not appear: ' + repr(stats))
        initial_frame, initial_stats = paint_frame('Browser-Monaco-Ready')
        details['MonacoReadyPaint'] = initial_stats
        start = len(text()); click(viewport[0] + 160, viewport[1] + 70)
        key('ctrl-end'); key('ret'); type_text('xfenativeedit')
        expected_edited = expected_initial + '\nxfenativeedit'
        end = time.monotonic() + 180
        while time.monotonic() < end:
            models = model_pattern.findall(healthy()[start:])
            if models and models[-1] == expected_edited: break
            time.sleep(.1)
        assert models and models[-1] == expected_edited, 'PS/2 input did not update the actual Monaco model exactly'
        details['MonacoEditedModel'] = models[-1]
        move(left + width - 20, top + 50)
        _, edited_stats = paint_frame('Browser-Monaco-Edited', True, initial_frame)
        details['MonacoEditedPaint'] = edited_stats
        details['MonacoKeyboardEditedModel'] = True
        details['Screenshots']['Monaco'] = screenshot('Browser-Monaco')
        details['OfficialMonacoReady'] = True
    # Keep the actual event loop alive after input and verify it responds again.
    time.sleep(2)
    start = len(text()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', start)
    screenshot('Desktop-After-Browser-Close')
    if documents_required:
        details['DocumentWorkflowProof'] = run_documents(proof, text, healthy, wait, screenshot,
            key, type_text, move, click, wheel)
    _, files_bounds = window_geometry(healthy(), 'Files')
    managed_row = seed_documents(True)['Names'].index('ManagedHello.dll')
    start = len(text()); click(files_bounds[0] + 70, files_bounds[1] + 110 + managed_row * 28)
    wait('XFE_INDEPENDENT_CONSOLE_IL_OK', start)
    wait('XFE_DESKTOP_ACTIVE|Terminal', start)
    wait('XFE_INDEPENDENT_CONSOLE_PRESENTED|Path|X:\\Users\\Default\\Documents\\ManagedHello.dll|Window|', start)
    details['ActualConsoleILSamplePassed'] = True
    details['Screenshots']['ConsoleIL'] = screenshot('Terminal-ConsoleIL')
    # Real terminal command dispatch replaces the removed demo-only toolbar.
    # The sample itself must produce its two actual ANSI/XFEConsole markers;
    # merely accepting a queued command cannot qualify program execution.
    start = len(text()); type_text('run ManagedHello.dll'); key('ret')
    wait('XFE_INDEPENDENT_CONSOLE_IL_OK', start)
    wait('XFE_INDEPENDENT_CONSOLE_PRESENTED|Path|X:\\Users\\Default\\Documents\\ManagedHello.dll|Window|', start)
    details['TerminalRunManagedSamplePassed'] = True
    details['ActualAnsiAndXfeConsoleDemoPassed'] = True
    details['Screenshots']['ConsoleDemo'] = screenshot('Terminal-Console-Demo')
    details['TerminalWorkflowProof'] = run_terminal_commands(proof, text, healthy, screenshot, key, type_text, input_kind)
    assert details['TerminalWorkflowProof']['Screenshot']['Sha256'] != details['Screenshots']['ConsoleDemo'], 'Actual CLI commands produced no changed guest framebuffer'
    healthy()
    details['NoRenderFailures'] = True
    return details


def verify_boot(kernel_path, cpus, serial_enabled, runtime_required, timeout, desktop_required=False, browser_required=False, documents_required=False, bing_required=False,
                console_required=False, existing_image=None, boot_entry=None, kernel_image_path=None, boot_wait_diagnostic=False):
    build = read_json(str(kernel_path) + '.json')
    assert build['Kind'] == 'XfeIndependentKernelBuild' and build['Passed'] is True
    assert build['ContainsCosmos'] is False and build['KernelSha256'] == digest(kernel_path)
    assert build['Target'] == 'x86_64-none'
    if runtime_required:
        assert build['BootstrapOnly'] is False, 'A bootstrap build cannot pass managed-runtime acceptance'
        assert serial_enabled, 'Runtime checks need independent serial evidence'
    if desktop_required:
        assert not build['BootstrapOnly'] and serial_enabled, 'Actual managed desktop acceptance requires a full build and serial evidence'
        assert not runtime_required, 'Runtime-probe acceptance and managed-application acceptance target separate entry points'
    edition = build.get('Edition', 'Desktop')
    if boot_wait_diagnostic:
        assert build['BootstrapOnly'] and serial_enabled and not any((desktop_required, console_required, runtime_required)), 'Waiting visual diagnostics require the native-only build, with no application acceptance'
    if console_required:
        assert edition == 'Console' and not build['BootstrapOnly'] and serial_enabled, 'Actual managed console acceptance requires a Console build and serial evidence'
        assert not runtime_required and not desktop_required and not browser_required and not documents_required and not bing_required, 'Console acceptance cannot qualify desktop/browser/runtime-probe entry points'
    if desktop_required:
        assert edition == 'Desktop', 'Desktop acceptance requires the Desktop edition'
    assert boot_entry is None or existing_image is not None, 'Explicit menu selection requires an existing dual image'
    assert kernel_image_path is None or existing_image is not None, 'An image kernel path applies only to an existing image'
    assert existing_image is None or serial_enabled, 'Dual menu acceptance requires serial evidence of the actual entry'
    selected_kernel = kernel_image_path or 'System/Kernel.elf'
    for relative, expected in build['SourceSha256'].items():
        path = (ROOT / relative).resolve()
        assert path.is_relative_to(ROOT) and digest(path) == expected, f'Source changed: {relative}'
    proof = ROOT / 'build' / 'IndependentKernel' / f'Proof-{time.time_ns()}'
    proof.mkdir(parents=True)
    visual_required = (desktop_required or console_required) and 'Source/XFESystem.NativeKernel/BootSplash.cs' in build['SourceSha256']
    boot_visual = BootAnimationCapture(proof)
    image = Path(existing_image).resolve() if existing_image is not None else proof / 'System.img'
    result = {'Kind': 'XfeIndependentKernelBootProof', 'Passed': False, 'ContainsCosmos': False,
              'Edition': edition,
              'VirtualCpus': cpus, 'ExecutingCpus': 1, 'SerialDevice': serial_enabled,
              'RuntimeAcceptanceRequired': runtime_required, 'DesktopAcceptanceRequired': desktop_required,
              'BrowserAcceptanceRequired': browser_required, 'ManagedRuntimeAcceptancePassed': False,
              'DocumentAcceptanceRequired': documents_required, 'DocumentAcceptancePassed': False,
              'ConsoleAcceptanceRequired': console_required, 'ConsoleAcceptancePassed': False,
              'BootVisualAcceptanceRequired': visual_required, 'BootVisualAcceptancePassed': False,
              'BootWaitVisualDiagnosticRequired': boot_wait_diagnostic, 'BootWaitVisualDiagnosticPassed': False,
              'BingAcceptanceRequired': bing_required, 'BingAcceptancePassed': False,
              'NativeNetworkProfile': native_network_profile(bing_required),
              'DesktopAcceptancePassed': False, 'BrowserAcceptancePassed': False,
              'PhysicalHardwareAcceptancePassed': False, 'FullReleaseQualified': False,
              'KernelSha256': digest(kernel_path), 'BuildReceipt': str(kernel_path) + '.json',
              'BuildReceiptSha256': digest(str(kernel_path) + '.json'),
              'RamOnly': True, 'StorageEnabled': False, 'ProofDirectory': str(proof)}
    qemu = None
    qmp = None
    sources = None
    try:
        if existing_image is None:
            with (proof / 'PackImage.log').open('w', encoding='utf-8') as pack_log:
                subprocess.run(['dotnet', 'run', '--project', str(ROOT / 'Tools/CSharpBoot/PackImage.csproj'),
                        '-c', 'Release', '--no-build', '--', str(kernel_path), str(image),
                        '1280x800', '--independent-kernel'], cwd=ROOT, check=True,
                       stdout=pack_log)
        inspected, files = inspect_image(image)
        assert files[selected_kernel] == kernel_path.read_bytes(), 'Selected image entry differs from the exact kernel under test'
        manifest = read_json(str(image) + '.json')
        assert manifest['KernelImplementation'] == 'XFESystem independent kernel'
        if existing_image is not None:
            assert manifest.get('Edition') == 'Dual' and boot_entry in (0, 1), 'Existing image acceptance requires an explicit dual menu selection'
            expected_entry = 1 if console_required else 0
            assert boot_entry == expected_entry and selected_kernel == ('System/Console.elf' if console_required else 'System/Kernel.elf'), 'Selected edition, menu index and kernel disagree'
            expected_build = manifest.get('IndependentConsoleKernelBuild' if console_required else 'IndependentKernelBuild')
            assert expected_build == build, 'The image selected build receipt differs from the exact compiled kernel'
        result.update(ImagePath=str(image), ImageSha256=digest(image), ImageVerification=inspected,
                      ImageManifestSha256=digest(str(image) + '.json'), KernelImagePath=selected_kernel,
                      BootEntry=boot_entry, BootShimSha256=inspected['Files']['EFI/BOOT/BOOTX64.EFI'])
        sources = dict(build['SourceSha256'])
        sources.update(inspected['BootShimSourceSha256'])
        for harness in ('Tools/IndependentKernel/TestBoot.py', 'Tools/IndependentKernel/IndependentDocumentEvidence.py',
                        'Tools/IndependentKernel/GpuDiscoveryEvidence.py',
                        'Tools/IndependentKernel/TestUsbInput.py',
                        'Tools/IndependentKernel/IndependentBingEvidence.py',
                        'Tools/IndependentKernel/IndependentConsoleEvidence.py',
                        'Tools/IndependentKernel/BootAnimationEvidence.py',
                        'Tools/IndependentKernel/VerifyDualDelivery.py', 'scripts/physical-media-verify.py',
                        'Tools/CSharpBoot/Program.cs', 'Tools/CSharpBoot/ReadImage.py',
                        'Tools/TestBrowserPipeline.py', 'Tools/TestDesktopDocuments.py', 'Tools/TestWebsiteDocument.py',
                        'Tools/NativeCSharpEvidence.py'):
            sources[harness] = digest(ROOT / harness)
        snapshot = proof / 'SourceSnapshot.json'
        snapshot.write_text(json.dumps(sources, indent=2), encoding='utf-8')
        result.update(SourceSnapshotPath=str(snapshot), SourceSnapshotSha256=digest(snapshot))
        shutil.copy2(ROOT / '.tools/firmware/edk2-i386-vars.fd', proof / 'FirmwareVars.fd')
        with socket.socket() as reservation:
            reservation.bind(('127.0.0.1', 0))
            port = reservation.getsockname()[1]
        command = [str(ROOT / '.tools/qemu/qemu-system-x86_64.exe'), '-machine', 'q35',
                   '-accel', 'tcg', '-cpu', 'max', '-smp', str(cpus), '-m', '1024M',
                   '-vga', 'std', '-global', 'VGA.vgamem_mb=64', '-display', 'none', *network_arguments(bing_required), '-rtc', 'base=utc',
                   '-drive', f'if=pflash,format=raw,readonly=on,file={ROOT / ".tools/firmware/edk2-x86_64-code.fd"}',
                   '-drive', f'if=pflash,format=raw,file={proof / "FirmwareVars.fd"}',
                   '-drive', f'file={image},format=raw,if=ide,snapshot=on',
                   '-serial', f'file:{proof / "Serial.log"}' if serial_enabled else 'none',
                   '-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off', '-no-reboot', '-no-shutdown']
        assert validate_network_arguments(command, bing_required) == result['NativeNetworkProfile']
        result['QemuArguments'] = command
        with (proof / 'Qemu.log').open('w', encoding='utf-8') as log:
            qemu = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
                                    creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
            deadline = time.monotonic() + timeout
            while qmp is None:
                assert qemu.poll() is None, 'QEMU exited before QMP was available'
                try:
                    qmp = Qmp(port)
                except (OSError, ConnectionError):
                    assert time.monotonic() < deadline, 'QMP connection timeout'
                    time.sleep(0.1)
            ready = False
            serial = ''
            frame = proof / 'Frame.ppm'
            menu_selected = boot_entry is None
            while time.monotonic() < deadline:
                assert qemu.poll() is None, 'Guest/QEMU exited during boot'
                if serial_enabled:
                    serial = (proof / 'Serial.log').read_text(encoding='utf-8', errors='replace')
                    assert not any(marker in serial for marker in ('XFE_NATIVE_FAIL', 'XFE_NATIVE_EXCEPTION', 'XFE_RUNTIME_FATAL',
                                                                   'XFE_ZERO_FAILFAST', 'XFE_INDEPENDENT_APPLICATION_FAILED')), serial[-4096:]
                    if not menu_selected:
                        # Limine's serial mirror supplies an actual menu-ready
                        # acknowledgement before sending physical keyboard keys.
                        if boot_menu_labels_present(serial):
                            menu_frame = proof / 'BootMenu.ppm'
                            qmp.call('screendump', {'filename': str(menu_frame)})
                            with Image.open(menu_frame) as rendered_menu:
                                rendered_menu.convert('RGB').save(proof / 'BootMenu.png')
                            for chord in ['home'] + ['down'] * boot_entry + ['ret']:
                                qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord})
                                time.sleep(.15)
                            result['BootMenuSelection'] = {'Input': 'Actual QMP PS2 keyboard packets only', 'Index': boot_entry,
                                'Entry': 'XFESystem Console' if console_required else 'XFESystem Desktop', 'MenuSerialOffset': len(serial),
                                'Screenshot': {'Path': 'BootMenu.png', 'Sha256': digest(proof / 'BootMenu.png')}}
                            menu_selected = True
                        elif 'XFE_NATIVE_SELFTEST_OK' in serial:
                            raise AssertionError('Guest booted before the requested dual menu entry was selected')
                    ready = 'XFE_NATIVE_SELFTEST_OK' in serial
                    if runtime_required:
                        ready = ready and 'XFE_DOTNET_RUNTIME_PROBE_OK' in serial
                    if desktop_required:
                        ready = ready and 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME' in serial and 'XFE_APPLICATION_RUNTIME_CHECK_OK' in serial
                    if console_required:
                        ready = ready and 'XFE_INDEPENDENT_CONSOLE_FIRST_FRAME' in serial and 'XFE_APPLICATION_RUNTIME_CHECK_OK' in serial
                        if 'XFE_SYSTEM_EDITION|Console' in serial: assert_console_isolation(serial)
                    ready = ready and menu_selected
                    if boot_wait_diagnostic:
                        ready = ready and 'XFE_BOOT_VISUAL_WAITING' in serial
                    if visual_required or boot_wait_diagnostic:
                        boot_visual.capture(qmp, serial)
                    if boot_wait_diagnostic:
                        ready = ready and any(item['Phase'] == 'Waiting' and item['LoaderPixels'] > 3 for item in boot_visual.frames)
                else:
                    qmp.call('screendump', {'filename': str(frame)})
                    with Image.open(frame) as rendered:
                        rgb = rendered.convert('RGB')
                        # Both actual application interfaces fill the screen;
                        # black/central splash frames cannot satisfy this.
                        ready = rgb.size == (1280, 800) and all(rgb.getpixel(point) != (0, 0, 0)
                                                               for point in ((10, 10), (10, 790), (1270, 790)))
                if ready:
                    break
                time.sleep(0.2)
            assert ready, 'Independent kernel boot/self-test timeout: ' + serial[-4096:]
            if serial_enabled:
                for marker in ('XFE_CPU_OK', 'XFE_MEMORY_MAP_OK', 'XFE_PAGES_OK', 'XFE_LIBC_OK',
                               'XFE_STDIO_SELFTEST_OK', 'XFE_LOADED_ELF_SELFTEST_OK', 'XFE_THREADS_OK', 'XFE_INPUT_READY',
                               'XFE_DESCRIPTORS_OK', 'XFE_VIRTUAL_MEMORY_OK', 'XFE_CLOCK_OK',
                               'XFE_FRAMEBUFFER_OK', 'XFE_NATIVE_SELFTEST_OK'):
                    assert marker in serial, marker
                result['GuestMarkers'] = [line for line in serial.splitlines() if line.startswith('XFE_')]
                gpu = validate_discovery(serial, expect_absent=True, require_firmware_handoff=True)
                probes = [line for line in serial.splitlines() if line.startswith('XFE_NVIDIA_BOOT_PROBE|')]
                assert len(probes) == 1 and '|State|1|' in probes[0], 'No-GPU guest must decline the physical GB202 register probe'
                assert '|GspStarted|0|ChannelCreated|0|Cuda|0|GpuRasterization|0' in probes[0]
                gpu.update(ReadOnlyBootProbe=probes[0], ActualNvidiaDeviceExecution=False)
                gpu_path = proof / 'GpuDiscoveryEvidence.json'
                gpu_path.write_text(json.dumps(gpu, indent=2), encoding='utf-8')
                result['GpuDiscoveryEvidence'] = {'Path': str(gpu_path), 'Sha256': digest(gpu_path),
                    'Passed': True, 'GuestCudaVerified': False, 'GpuRasterizationVerified': False}
            if visual_required:
                result['BootVisualProof'] = boot_visual.finish(serial,
                    'XFE_INDEPENDENT_CONSOLE_FIRST_FRAME' if console_required else 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME')
                result['BootVisualAcceptancePassed'] = True
            if boot_wait_diagnostic:
                result['BootWaitVisualProof'] = boot_visual.finish_waiting_diagnostic(serial)
                result['BootWaitVisualDiagnosticPassed'] = True

            qmp.call('screendump', {'filename': str(frame)})
            with Image.open(frame) as rendered:
                rgb = rendered.convert('RGB')
                assert rgb.size == (1280, 800)
                rgb.save(proof / 'Frame.png')
                result['ScreenshotSha256'] = digest(proof / 'Frame.png')
                result['FramebufferSize'] = list(rgb.size)
            qmp.call('query-status')
            if desktop_required:
                result['DesktopInputProof'] = exercise_desktop(qmp, proof,
                    lambda: (proof / 'Serial.log').read_text(encoding='utf-8', errors='replace'), browser_required, documents_required, bing_required)
                result['DesktopAcceptancePassed'] = True
                result['BrowserAcceptancePassed'] = browser_required
                result['DocumentAcceptancePassed'] = documents_required
                result['BingAcceptancePassed'] = bing_required and result['DesktopInputProof']['BingEvidence']['Passed']
            if console_required:
                result['ConsoleInputProof'] = run_console(qmp, proof,
                    lambda: (proof / 'Serial.log').read_text(encoding='utf-8', errors='replace'))
                result['ConsoleAcceptancePassed'] = result['ConsoleInputProof']['Passed']
            result['ManagedRuntimeAcceptancePassed'] = runtime_required or desktop_required or console_required
            result['Passed'] = True
    except Exception as error:
        result['Failure'] = str(error)
    finally:
        if qemu is not None:
            if qemu.poll() is None and qmp is not None:
                try:
                    qmp.call('quit')
                except (OSError, RuntimeError):
                    pass
            try:
                qemu.wait(timeout=10)
            except subprocess.TimeoutExpired:
                qemu.terminate()
                try:
                    qemu.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    qemu.kill(); qemu.wait(timeout=10)
                result['ForcedCleanup'] = True
            result['QemuExitCode'] = qemu.returncode
        if qmp is not None:
            qmp.close()
        if image.exists():
            result['ImageUnchanged'] = digest(image) == result.get('ImageSha256')
        serial_path = proof / 'Serial.log'
        if serial_enabled and serial_path.exists():
            result['SerialEvidence'] = {'Path': serial_path.name, 'Sha256': digest(serial_path)}
        if console_required and serial_path.exists():
            final_serial = serial_path.read_text(encoding='utf-8', errors='replace')
            try:
                assert_console_isolation(final_serial)
                result['NoDesktopOrBrowserInitialization'] = True
            except AssertionError as error:
                result['Passed'] = False; result['Failure'] = str(error)
        if sources is not None:
            try:
                for relative, expected in sources.items():
                    path = (ROOT / relative).resolve()
                    assert path.is_relative_to(ROOT) and digest(path) == expected, 'Source changed during guest acceptance: ' + relative
                assert digest(result['SourceSnapshotPath']) == result['SourceSnapshotSha256'], 'Guest source snapshot changed during acceptance'
                assert digest(kernel_path) == result['KernelSha256'] and digest(str(kernel_path) + '.json') == result['BuildReceiptSha256'], 'Compiled kernel/build receipt changed during acceptance'
                assert digest(str(image) + '.json') == result['ImageManifestSha256'], 'Image manifest changed during guest acceptance'
                result['SourcesUnchanged'] = True
            except (AssertionError, OSError) as error:
                result['Passed'] = False; result['SourcesUnchanged'] = False; result['Failure'] = str(error)
        result['Passed'] = result['Passed'] and result.get('QemuExitCode') == 0 and result.get('ImageUnchanged') is True and result.get('SourcesUnchanged') is True
        (proof / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
    print(json.dumps(result, ensure_ascii=True))
    return result['Passed']


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('kernel', type=Path)
    parser.add_argument('--virtual-cpus', type=int, choices=(1, 8), default=1)
    parser.add_argument('--no-serial', action='store_true')
    parser.add_argument('--runtime', action='store_true')
    parser.add_argument('--desktop', action='store_true')
    parser.add_argument('--browser', action='store_true')
    parser.add_argument('--documents', action='store_true')
    parser.add_argument('--console', action='store_true', help='Require actual full-screen Console edition RAM shell, history, caret editing and embedded managed sample')
    parser.add_argument('--boot-visual-wait', action='store_true', help='Native-only diagnostic: show actual loading pixels and never release an unready application')
    parser.add_argument('--image', type=Path, help='Verify an existing dual image without repacking it')
    parser.add_argument('--boot-entry', type=int, choices=(0, 1), help='Actual Limine menu index selected through QMP PS/2 input')
    parser.add_argument('--kernel-image-path', choices=('System/Kernel.elf', 'System/Console.elf'))
    parser.add_argument('--bing', action='store_true', help='Require real guest e1000/NAT HTTPS Bing, immutable snapshot presentation and actual framebuffer evidence')
    parser.add_argument('--timeout', type=int, default=90)
    args = parser.parse_args()
    raise SystemExit(0 if verify_boot(args.kernel.resolve(), args.virtual_cpus, not args.no_serial,
                                     args.runtime, args.timeout, args.desktop or args.browser or args.documents or args.bing,
                                     args.browser or args.documents, args.documents, args.bing, args.console,
                                     args.image, args.boot_entry, args.kernel_image_path, args.boot_visual_wait) else 1)