XFE Git
XFE Studio Git
Git 首页 全局搜索
XFE 主站 文档 NuGet

XFESystem

【Rust】XFE系统,基于Rust和汇编

公开
关注 0 Fork 0 Star 0
UTF-8
"""Boot the real NativeAOT ELF under QEMU/UEFI; never runs the kernel on Windows."""
import hashlib
import argparse
import json
import pathlib
import re
import shutil
import socket
import subprocess
import sys
import time
import uuid
from PIL import Image
from ReadImage import inspect_image

# Keep the CLI receipt UTF-8 when Windows redirects stdout using a legacy
# locale. Website/KaTeX readback legitimately contains zero-width Unicode.
for output_stream in (sys.stdout, sys.stderr):
    if hasattr(output_stream, 'reconfigure'):
        output_stream.reconfigure(encoding='utf-8')


def bounded_failure_stack(registers, read_hmp):
    """Read only [RSP,RSP+32KiB); walk RBP using these already read words."""
    rsp_match = re.search(r'\bRSP=(?:0x)?([0-9a-fA-F]{1,16})\b', registers)
    assert rsp_match, 'The private guest register report has no x86_64 RSP'
    rsp = int(rsp_match[1], 16)
    byte_count = 32 * 1024
    assert rsp % 8 == 0 and 0 < rsp <= (1 << 64) - byte_count, 'Invalid bounded stack geometry'
    end = rsp + byte_count
    command = f'x/4096gx {rsp:#x}'
    dump = read_hmp(command)
    words = {}
    for line in dump.splitlines():
        row = re.match(r'^\s*(?:0x)?([0-9a-fA-F]{1,16}):\s*(.*)$', line)
        if row is None:
            continue
        address = int(row[1], 16)
        for index, value in enumerate(re.findall(r'0x([0-9a-fA-F]{1,16})\b', row[2])):
            at = address + index * 8
            if rsp <= at < end and (at - rsp) % 8 == 0:
                words[at] = int(value, 16)
    rbp_match = re.search(r'\bRBP=(?:0x)?([0-9a-fA-F]{1,16})\b', registers)
    rip_match = re.search(r'\bRIP=(?:0x)?([0-9a-fA-F]{1,16})\b', registers)
    frame = int(rbp_match[1], 16) if rbp_match else 0
    frames = []
    stop = 'MissingRbp' if rbp_match is None else 'FrameLimit'
    for _ in range(128):
        if frame == 0:
            stop = 'NullFramePointer'; break
        if not (rsp <= frame <= end - 16) or frame % 8 != 0:
            stop = 'OutsideBoundedStackRange'; break
        if frame not in words or frame + 8 not in words:
            stop = 'FrameWordsUnreadable'; break
        next_frame, return_address = words[frame], words[frame + 8]
        frames.append({'FramePointer': frame, 'NextFramePointer': next_frame, 'ReturnAddress': return_address})
        if next_frame != 0 and next_frame <= frame:
            stop = 'NonIncreasingFramePointer'; break
        frame = next_frame
    return {'ReadOnly': True, 'StartInclusive': rsp, 'EndExclusive': end, 'MaximumBytes': byte_count,
        'HmpCommand': command, 'WordsRead': len(words), 'CompleteRead': len(words) == byte_count // 8,
        'InstructionPointer': int(rip_match[1], 16) if rip_match else None,
        'RbpFrames': frames, 'RbpStopReason': stop, 'RawDump': dump,
        'Meaning': 'Bounded raw virtual stack and frame-pointer evidence from this private guest; no memory write or injected guest code.'}


ROOT = pathlib.Path(__file__).resolve().parents[2]
parser = argparse.ArgumentParser()
parser.add_argument('--resolution', default='1280x800')
parser.add_argument('--smoke-display', action='store_true')
parser.add_argument('--desktop-ui', action='store_true')
parser.add_argument('--browser-pipeline', action='store_true', help='Real CSS, external script, input event and official Monaco guest checks')
parser.add_argument('--preview-performance', action='store_true', help='Actual Markdown Preview click, first paint and desktop responsiveness timings')
parser.add_argument('--document-first-paint', action='store_true', help='Actual original-CSS Markdown first presentation before enhancement scripts; report 1s objective separately')
parser.add_argument('--editor-responsiveness', action='store_true', help='Open actual C# Monaco while exercising desktop shortcuts during its largest script')
parser.add_argument('--desktop-documents', action='store_true', help='Markdown preview-first, asynchronous Edit/Split, save/reopen and honest Bing default-page readiness')
parser.add_argument('--pointer-cadence', action='store_true', help='Bounded actual pointer input/software-present timings and raw screenshots; does not qualify physical monitor scanout')
parser.add_argument('--usb-input', action='store_true', help='Disable PS/2 and use a unique xHCI USB keyboard and relative USB mouse')
parser.add_argument('--editor-stability', action='store_true', help='Sustained real Monaco input, save, website preview and reopen lifecycle checks')
parser.add_argument('--stability-idle-seconds', type=int, default=300, help='Focused editor idle observation duration for --editor-stability (default 300)')
parser.add_argument('--stability-cycles', type=int, default=12, help='C#/JS/Markdown edit-save-preview-close cycles (default 12; 0 runs only the C# idle anchor)')
parser.add_argument('--browser-oom-negative', action='store_true', help='Deliberate allocation failure; never counts as a passing interactive image')
parser.add_argument('--desktop-v4', action='store_true', help='No-topbar desktop, caption motion and native drag/resize acceptance')
parser.add_argument('--desktop-ui-only', action='store_true', help='Targeted window/performance profile; does not count baseline application/HTTP interactions as checked')
parser.add_argument('--accel', choices=['tcg', 'whpx'], default='tcg')
parser.add_argument('--whpx-userspace-irqchip', action='store_true',
                    help='WHPX-only diagnostic: keep CPU acceleration but use QEMU userspace APIC; requires separate native timing acceptance')
parser.add_argument('--virtual-cpus', type=int, choices=[1, 8], default=1,
                    help='Advertised QEMU/ACPI CPU count; the kernel starts only the BSP in its single-CPU execution model')
parser.add_argument('--network-device', choices=['e1000e', 'none'], default='e1000e',
                    help='Use none to verify bounded offline startup without a network adapter')
parser.add_argument('--display-backend', choices=['none', 'gtk', 'sdl'], default='none',
                    help='Actual QEMU display backend; SDL/GTK are visible host windows, never guest GPU/CUDA acceleration')
parser.add_argument('--input-rate', type=int, default=500)
parser.add_argument('--verified-image')
parser.add_argument('--verified-boot-result')
parser.add_argument('--evidence-root', help='Absolute evidence directory; fresh image packaging requires an in-repository path, verified-image reuse also supports external paths; existing runs are never deleted')
parser.add_argument('--boot-timeout', type=int, default=90)
parser.add_argument('--memory-mib', type=int, default=1024, choices=[512,1024,2048], help='Explicit memory profile; never inferred as 512 MiB acceptance')
parser.add_argument('--video-memory-mib', type=int, choices=[16, 32, 64], default=64)
parser.add_argument('--edid-resolution', default=None)
parser.add_argument('--refresh-hint', type=int, default=240, help='EDID timing hint in Hz; QEMU property is converted to millihertz')
options = parser.parse_args()
if options.whpx_userspace_irqchip and options.accel != 'whpx':
    parser.error('--whpx-userspace-irqchip requires --accel whpx; no VM or artifacts were created')
accelerator_argument = 'whpx,kernel-irqchip=off' if options.whpx_userspace_irqchip else options.accel
if options.editor_stability:
    assert 0 <= options.stability_idle_seconds <= 3600 and 0 <= options.stability_cycles <= 120
if options.evidence_root:
    evidence_root = pathlib.Path(options.evidence_root)
    assert evidence_root.is_absolute(), '--evidence-root must be an absolute directory'
    evidence_root = evidence_root.resolve()
else:
    evidence_root = ROOT / 'build' / 'CSharpBoot'
# Refuse another image/ELF copy before the host volume approaches exhaustion.
storage_parent = evidence_root
while not storage_parent.exists():
    storage_parent = storage_parent.parent
if shutil.disk_usage(storage_parent).free < 10 * 1024 ** 3:
    parser.error('At least 10 GiB of free disk space is required before a native test; no artifacts were created')
OUT = evidence_root / f'Proof-{time.time_ns()}'
OUT.mkdir(parents=True)
ELF = ROOT / 'Source/XFESystem.Kernel/bin/Release/net10.0/linux-x64/XFESystem.Kernel.elf'
DISK = OUT / 'System.img'
prior_receipt = None
if options.verified_image:
    assert options.verified_boot_result, 'A verified image requires its prior boot receipt'
    prior_receipt_path = pathlib.Path(options.verified_boot_result).resolve()
    prior_receipt = json.loads(prior_receipt_path.read_text(encoding='utf-8-sig'))
    assert prior_receipt['Passed'] and prior_receipt['ImageUnchanged'] and prior_receipt['QemuExitCode'] == 0
    source_disk = pathlib.Path(options.verified_image).resolve()
    assert hashlib.sha256(source_disk.read_bytes()).hexdigest() == prior_receipt['ImageSha256']
    shutil.copyfile(source_disk, DISK)
    options.resolution = prior_receipt['ResolutionRequested']
else:
    assert not options.verified_boot_result, 'A prior boot receipt requires its verified image'
    subprocess.run(['dotnet', 'run', '--project', str(ROOT / 'Tools/CSharpBoot/PackImage.csproj'), '-c', 'Release', '--', str(ELF), str(DISK), options.resolution], cwd=ROOT, check=True)
image_hash = hashlib.sha256(DISK.read_bytes()).hexdigest()
image_verification, image_files = inspect_image(DISK)
kernel_bytes = image_files['System/Kernel.elf']
if prior_receipt:
    assert hashlib.sha256(kernel_bytes).hexdigest() == prior_receipt['KernelSha256']
else:
    assert kernel_bytes == ELF.read_bytes()
# Historical receipts must reference their own exact image payload, never the
# mutable default compiler output that a subsequent build can replace.
ELF = OUT / 'Kernel.elf'
ELF.write_bytes(kernel_bytes)
shim_receipt = json.loads((ROOT / 'build/CSharpBoot/BootShimBuild/BootShimReceipt.json').read_text(encoding='utf-8-sig'))
assert image_files['EFI/BOOT/BOOTX64.EFI'] == pathlib.Path(shim_receipt['ArtifactPath']).read_bytes()
assert image_files['EFI/XFE/Limine.efi'] == (ROOT / '.tools/limine/BOOTX64.EFI').read_bytes()
assert b'path: boot():/System/Kernel.elf' in image_files['Limine.conf']
shutil.copy2(ROOT / '.tools/firmware/edk2-i386-vars.fd', OUT / 'FirmwareVars.fd')
with socket.socket() as probe:
    probe.bind(('127.0.0.1', 0))
    port = probe.getsockname()[1]
display_argument = {'none': 'none', 'gtk': 'gtk,gl=off', 'sdl': 'sdl,gl=off'}[options.display_backend]
guest_uuid = str(uuid.uuid4())
args = [str(ROOT / '.tools/qemu/qemu-system-x86_64.exe'), '-uuid', guest_uuid, '-machine', 'q35,i8042=off,usb=off' if options.usb_input else 'q35',
        '-accel', accelerator_argument, '-cpu', 'max', '-m', str(options.memory_mib)+'M', '-smp', str(options.virtual_cpus),
        '-vga', 'std', '-global', f'VGA.vgamem_mb={options.video_memory_mib}',
        '-drive', f'if=pflash,format=raw,readonly=on,file={ROOT / ".tools/firmware/edk2-x86_64-code.fd"}',
        '-drive', f'if=pflash,format=raw,file={OUT / "FirmwareVars.fd"}',
        '-drive', f'file={DISK},format=raw,if=ide,snapshot=on',
        '-display', display_argument, '-serial', f'file:{OUT / "Serial.log"}',
        '-monitor', 'none', '-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off',
        '-no-reboot']
if options.usb_input:
    args += ['-device', 'qemu-xhci,id=XfeInputXhci', '-device', 'usb-kbd,id=XfeKeyboard,bus=XfeInputXhci.0',
             '-device', 'usb-mouse,id=XfeMouse,bus=XfeInputXhci.0']
if options.network_device != 'none':
    args += ['-netdev', 'user,id=NativeNet', '-device', 'e1000e,netdev=NativeNet']
else:
    args += ['-nic', 'none']
edid_resolution = options.edid_resolution or ('3840x2160' if options.resolution == '0x0' else options.resolution)
edid_width, edid_height = (int(value) for value in edid_resolution.split('x'))
assert 320 <= edid_width <= 16384 and 200 <= edid_height <= 16384 and 20 <= options.refresh_hint <= 1000
args += ['-global', f'VGA.xres={edid_width}', '-global', f'VGA.yres={edid_height}',
         '-global', f'VGA.xmax={edid_width}', '-global', f'VGA.ymax={edid_height}',
         '-global', 'VGA.edid=on', '-global', f'VGA.refresh_rate={options.refresh_hint * 1000}']
result = {'Kind': 'BareMetalNativeAotBootProof', 'KernelSha256': hashlib.sha256(ELF.read_bytes()).hexdigest(),
          'KernelPath': str(ELF), 'ImagePath': str(DISK),
          'EvidenceRoot': str(evidence_root.resolve()), 'EvidenceRootExplicit': bool(options.evidence_root),
          'EvidenceDirectoryPolicy': 'Create a new Proof child only; existing evidence is never removed or overwritten',
          'Runtime': 'Cosmos NativeAOT plugs 3.0.94', 'FullDotNet10': False, 'DotNetFramework48': False,
          'ManagedConsoleIlProfile': 'XFESystem.Net10.ConsoleIL.v1',
          'ManagedSampleSha256': hashlib.sha256((ROOT / 'Samples/XFESystem.ManagedHello/bin/Release/net10.0/XFESystem.ManagedHello.dll').read_bytes()).hexdigest(),
          'ResolutionRequested': options.resolution,
          'EdidResolution': edid_resolution, 'EdidRefreshHintHz': options.refresh_hint, 'QemuRefreshRateMilliHz': options.refresh_hint * 1000,
          'RefreshMeaning': 'EDID target; actual presentation/VSync performance is measured separately',
          'Acceleration': options.accel, 'InputEventRateRequested': options.input_rate,
          'AcceleratorArgument': accelerator_argument,
          'WhpxKernelIrqchipRequested': 'off' if options.whpx_userspace_irqchip else 'default',
          'WhpxUserspaceIrqchip': options.whpx_userspace_irqchip,
          'WhpxIrqchipMode': 'UserspaceApic' if options.whpx_userspace_irqchip else ('QemuDefaultPolicy' if options.accel == 'whpx' else 'NotApplicable'),
          'DefaultIrqchipTimingAcceptanceReused': False,
          'WhpxIrqchipTimingEvidencePolicy': ('WHPX CPU acceleration with QEMU userspace APIC; default-irqchip timing acceptance is not reused. This VM configuration requires its own native boot, input and timing measurements.'
                                            if options.whpx_userspace_irqchip else 'No irqchip override; QEMU default accelerator policy is preserved.'),
          'DisplayBackend': options.display_backend, 'QemuDisplayArgument': display_argument,
          'VisibleDisplayRequested': options.display_backend != 'none', 'HostGlEnabled': False,
          'MemoryMiB': options.memory_mib,
          'VirtualCpuCount': options.virtual_cpus, 'KernelStartedCpuCount': 1,
          'UsbInputProfile': options.usb_input,
          'PrivateQemuUuid': guest_uuid,
          'NetworkDevice': options.network_device,
          'DesktopV4Profile': options.desktop_v4,
          'BootShimSha256': shim_receipt['ArtifactSha256'],
          'TestHarnessSha256': hashlib.sha256(pathlib.Path(__file__).read_bytes()).hexdigest(),
          'UnrestrictedManagedDllLoading': False, 'QemuArgs': args, 'Passed': False}
source_hashes = {'Directory.Build.targets': hashlib.sha256((ROOT / 'Directory.Build.targets').read_bytes()).hexdigest()}
for source_root in [ROOT / 'Source', ROOT / 'Samples', ROOT / 'Tools', ROOT / 'assets/DesktopIcons', ROOT / 'assets/Browser', ROOT / 'assets/fonts', ROOT / 'assets/branding/console']:
    for source in sorted(source_root.rglob('*')):
        if source.is_file() and not {'bin', 'obj', 'build', 'target', '__pycache__'}.intersection(source.relative_to(source_root).parts):
            source_hashes[source.relative_to(ROOT).as_posix()] = hashlib.sha256(source.read_bytes()).hexdigest()
source_snapshot = json.dumps(source_hashes, sort_keys=True, separators=(',', ':')).encode('utf-8')
(OUT / 'SourceSnapshot.json').write_bytes(source_snapshot)
result['SourceSnapshotSha256'] = hashlib.sha256(source_snapshot).hexdigest()
result['SourceSnapshotPath'] = str(OUT / 'SourceSnapshot.json')
if prior_receipt:
    current_snapshot = OUT / 'CurrentHarnessSourceSnapshot.json'
    (OUT / 'SourceSnapshot.json').rename(current_snapshot)
    result['CurrentHarnessSourceSnapshotSha256'] = result['SourceSnapshotSha256']
    result['CurrentHarnessSourceSnapshotPath'] = str(current_snapshot)
    previous_snapshot = pathlib.Path(prior_receipt['SourceSnapshotPath'])
    assert hashlib.sha256(previous_snapshot.read_bytes()).hexdigest() == prior_receipt['SourceSnapshotSha256']
    shutil.copyfile(previous_snapshot, OUT / 'SourceSnapshot.json')
    result['SourceSnapshotSha256'] = prior_receipt['SourceSnapshotSha256']
    result['SourceSnapshotBinding'] = 'Compiled-image source snapshot reused from the verified prior boot; current harness recorded separately'
    result['VerifiedPriorBootResult'] = str(prior_receipt_path)
shutil.copyfile(ROOT / 'build/CSharpBoot/BootShimBuild/BootShimReceipt.json', OUT / 'BootShimReceipt.json')
result['BootShimReceiptSha256'] = hashlib.sha256((OUT / 'BootShimReceipt.json').read_bytes()).hexdigest()
runtime_patches = {}
for name, filename, suffix in [('Input','Cosmos.Kernel.System.dll','.XfeInput.json'),('JintCleanup','Jint.dll','.XfeCleanup.json'),('GcTail','Cosmos.Kernel.Core.dll','.XfeGc.json')]:
    runtime_copy = ROOT / 'Source/XFESystem.Kernel/obj/Release/net10.0/linux-x64/cosmos/ref' / filename
    patch_source = pathlib.Path(str(runtime_copy)+suffix)
    patch = json.loads(patch_source.read_text(encoding='utf-8-sig'))
    assert patch['GlobalPackageCacheModified'] is False
    assert hashlib.sha256(runtime_copy.read_bytes()).hexdigest() == patch['AfterSha256']
    if prior_receipt:
        assert prior_receipt['PrivateRuntimePatches'][name]['RuntimeSha256'] == patch['AfterSha256'], 'A reused image cannot be rebound to a different current runtime patch'
    if name == 'GcTail':
        assert patch['SchedulerForcedTimesliceNanoseconds'] == 1_000_000
        assert patch['SchedulerStrideAccountingDivisorNanoseconds'] == 10_000_000
        assert patch['SchedulerForcedTimesliceAdjusted'] is True and patch['SchedulerPassBasedPreemptionPreserved'] is True
        assert patch['SchedulerTimesliceChangedInstructionIndex'] == 103
    shutil.copyfile(patch_source,OUT/(name+'RuntimePatch.json'))
    runtime_patches[name] = {'Path':str(OUT/(name+'RuntimePatch.json')),'Sha256':hashlib.sha256(patch_source.read_bytes()).hexdigest(),'RuntimeSha256':patch['AfterSha256']}
    if name == 'GcTail':
        runtime_patches[name].update({'SchedulerForcedTimesliceNanoseconds': patch['SchedulerForcedTimesliceNanoseconds'],
                                     'SchedulerStrideAccountingDivisorNanoseconds': patch['SchedulerStrideAccountingDivisorNanoseconds']})
result['PrivateRuntimePatches'] = runtime_patches
platform_source = ROOT / 'Source/XFESystem.Kernel/obj/Release/net10.0/linux-x64/cosmos/ref/XfeBootPlatformV1.json'
if platform_source.exists():
    assert not platform_source.with_name('XfeBootPlatformV1.Pending.json').exists(), 'Partial platform patch transaction'
    platform_patch = json.loads(platform_source.read_text(encoding='utf-8-sig'))
    assert platform_patch['Kind'] == 'ProjectLocalBootProcessorPlatformV1' and platform_patch['PackageVersion'] == '3.0.94'
    assert platform_patch['SchedulerCpuCount'] == 1 and platform_patch['NormalTimerLogRemoved'] is True
    assert platform_patch['HardwareApicRoutingPreserved'] is True
    assert platform_patch['DeviceIrqSchedulerUsesLogicalCpuZero'] is True
    assert platform_patch['SchedulerCpuIndex'] == 0 and platform_patch['SchedulerCpuIndexSitesPatched'] == 2
    for filename, field in [('Cosmos.Kernel.Core.X64.dll', 'CoreAfterSha256'), ('Cosmos.Kernel.HAL.X64.dll', 'HalAfterSha256'),
                            ('Cosmos.Kernel.Drivers.dll', 'DriversAfterSha256'), ('System.Private.CoreLib.dll', 'CoreLibAfterSha256')]:
        assert hashlib.sha256((platform_source.parent / filename).read_bytes()).hexdigest() == platform_patch[field]
    assert platform_patch['SchedulerTimerIntervalMs'] == 1 and platform_patch['XhciPolledDrainIntervalMs'] == 1
    assert platform_patch['SuppressedMonitorLogCalls'] == 9 and platform_patch['MonitorWaitAlgorithmsPreserved']
    if prior_receipt:
        prior_platform = prior_receipt['PrivateRuntimePatches']['BootPlatform']
        for previous_field, current_field in [('CoreRuntimeSha256', 'CoreAfterSha256'), ('HalRuntimeSha256', 'HalAfterSha256'),
                                               ('DriversRuntimeSha256', 'DriversAfterSha256'), ('CoreLibRuntimeSha256', 'CoreLibAfterSha256')]:
            assert prior_platform[previous_field] == platform_patch[current_field], 'A reused image has a different bound platform patch'
    shutil.copyfile(platform_source, OUT / 'BootPlatformRuntimePatch.json')
    result['PrivateRuntimePatches']['BootPlatform'] = {'Path': str(OUT / 'BootPlatformRuntimePatch.json'),
        'Sha256': hashlib.sha256(platform_source.read_bytes()).hexdigest(),
        'CoreRuntimeSha256': platform_patch['CoreAfterSha256'], 'HalRuntimeSha256': platform_patch['HalAfterSha256'],
        'DriversRuntimeSha256': platform_patch['DriversAfterSha256'], 'CoreLibRuntimeSha256': platform_patch['CoreLibAfterSha256'],
        'SchedulerTimerIntervalMs': platform_patch['SchedulerTimerIntervalMs'],
        'XhciPolledDrainIntervalMs': platform_patch['XhciPolledDrainIntervalMs']}
elif not prior_receipt:
    raise AssertionError('A fresh current image requires its verified single-CPU platform patch receipt')
process = subprocess.Popen(args, cwd=ROOT, stdout=(OUT / 'QemuStdout.log').open('wb'), stderr=(OUT / 'QemuStderr.log').open('wb'))
client = None
fixture = None
def text():
    return (OUT / 'Serial.log').read_text(encoding='utf-8', errors='replace') if (OUT / 'Serial.log').exists() else ''

def desktop_snapshots(log):
    """Read complete atomic guest state reports, never historical per-window rows.

    The first subsequent non-state line (or the next FLAGS header) seals a
    report. An EOF after a row can still be a partially delivered serial write,
    so it cannot prove that a later window row is absent.
    """
    headers = list(re.finditer(r'^XFE_DESKTOP_FLAGS\|([01])\|([01])\|Active\|(\d+)\r?\n', log, re.M))
    row_pattern = re.compile(r'XFE_WINDOW_STATE\|(\d+)\|([^|]+)\|(-?\d+)\|(-?\d+)\|(\d+)\|(\d+)\|([01])\|([01])\|([01])')
    snapshots = []
    for index, header in enumerate(headers):
        end = headers[index + 1].start() if index + 1 < len(headers) else len(log)
        section = log[header.end():end]
        windows = {}
        sealed = index + 1 < len(headers)
        appearance_seen = False
        for line in section.splitlines(keepends=True):
            if not line.endswith('\n'):
                break
            line = line.rstrip('\r\n')
            if not appearance_seen:
                if not line.startswith('XFE_DESKTOP_APPEARANCE|'):
                    break
                appearance_seen = True
                continue
            row = row_pattern.fullmatch(line)
            if row is None:
                sealed = True
                break
            values = row.groups()
            windows[int(values[0])] = {'Application': values[1], 'Bounds': [int(value) for value in values[2:6]],
                                       'Minimized': int(values[6]), 'Maximized': int(values[7]), 'Snapped': int(values[8])}
        if appearance_seen and sealed:
            snapshots.append({'ActiveId': int(header.group(3)), 'StartMenuOpen': int(header.group(1)),
                              'WindowSwitching': int(header.group(2)), 'Windows': windows, 'SerialOffset': header.start()})
    return snapshots

try:
    deadline = time.monotonic() + options.boot_timeout
    while time.monotonic() < deadline:
        if process.poll() is not None:
            raise RuntimeError(f'QEMU exited {process.returncode}')
        log = (OUT / 'Serial.log').read_text(encoding='utf-8', errors='replace') if (OUT / 'Serial.log').exists() else ''
        if '[FAILFAST]' in log or '*** UNHANDLED EXCEPTION ***' in log:
            raise RuntimeError('Guest runtime failure; inspect Serial.log')
        if 'XFE_CSHARP_DESKTOP_READY' in log:
            result['Booted'] = True
            break
        time.sleep(.2)
    if not result.get('Booted'):
        raise RuntimeError('Guest boot marker timeout; inspect Serial.log')
    if options.display_backend != 'none':
        assert sys.platform == 'win32', 'Visible-window observation currently requires the supported Windows host'
        window_script = (f'$p=Get-Process -Id {process.pid} -ErrorAction Stop; '
                         '[pscustomobject]@{Handle=$p.MainWindowHandle.ToInt64();Title=$p.MainWindowTitle} | ConvertTo-Json -Compress')
        window_probe = subprocess.run(['powershell.exe', '-NoProfile', '-Command', window_script],
                                      capture_output=True, text=True, encoding='utf-8', check=True)
        window = json.loads(window_probe.stdout)
        result['VisibleWindow'] = {'Observed': window['Handle'] != 0, 'ProcessId': process.pid,
                                   'MainWindowHandle': window['Handle'], 'Title': window['Title'],
                                   'Backend': options.display_backend, 'HostGlEnabled': False}
        assert result['VisibleWindow']['Observed'], 'No actual QEMU display window was observed'
    time.sleep(1)
    client = socket.create_connection(('127.0.0.1', port), timeout=5)
    stream = client.makefile('rwb')
    stream.readline()
    def qmp(command, arguments=None):
        stream.write((json.dumps({'execute': command, 'arguments': arguments or {}}) + '\n').encode())
        stream.flush()
        while True:
            message = json.loads(stream.readline())
            if 'return' in message:
                return message['return']
            if 'error' in message:
                raise RuntimeError(message['error'])
    qmp('qmp_capabilities')
    assert qmp('query-uuid')['UUID'].lower() == guest_uuid, 'QMP is not this harness-owned guest'
    def wait_log(marker, start=0, timeout=10):
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            current_log = text()
            if marker in current_log[start:]:
                return
            if any(failure_marker in current_log for failure_marker in
                   ('CPU EXCEPTION:', '[FAILFAST]', 'XFE_RUNTIME_OOM|', 'XFE_RUNTIME_STACK_OOM|')):
                raise RuntimeError('Guest exception before ' + marker + ': ' + current_log[-4000:])
            if process.poll() is not None:
                raise RuntimeError(f'Guest exited before {marker}')
            time.sleep(.05)
        raise RuntimeError(f'Guest input/output timeout: {marker}')
    def screenshot(name, convert=True):
        ppm = OUT / f'{name}.ppm'
        qmp('screendump', {'filename': str(ppm)})
        if not convert: return ppm
        png = OUT / f'{name}.png'
        with Image.open(ppm) as frame:
            frame.save(png)
        # A 4K raw screenshot costs about 25 MB. Keep only its lossless PNG
        # after conversion succeeds; failed conversions retain their source.
        ppm.unlink()
        return png
    for marker in ['XFE_CSHARP_NATIVEAOT_BOOT_OK', 'XFE_RUNTIME_VARIANCE_OK', 'XFE_CSHARP_CONSOLE_DEMO_OK',
                   'XFE_GC_DEEP_RESULT|True|Frames|321|Collections|2|FailureDepth|-1',
                   'XFE_EH_DEEP_RETHROW_OK|Frames|81|Post|4',
                   'XFE_THREAD_EH_GC_RESULT|True|Steps|32|Collections|4|SingleCpu|1|GcStopTheWorld|1',
                   'XFE_BROWSER_EVENT_REGRESSION|Complete|PASS',
                   'XFE_CSHARP_ANSI_RGB_OK', 'XFE_CSHARP_CONSOLE_PAL_COLOR_OK',
                   'XFE_CSHARP_XFECONSOLE_API_OK', 'XFE_MANAGED_APP_OK',
                   'XFE_MANAGED_XFECONSOLE_OK', 'XFE_MANAGED_ANSI_RGB_OK',
                   'XFE_JAVASCRIPT_LIMIT_RECOVERY_OK',
                   'XFE_CSHARP_JAVASCRIPT_DOM_OK', 'XFE_CSHARP_MARKDOWN_OK']:
        wait_log(marker)
    preemption = re.search(r'XFE_THREAD_PREEMPTION\|Checks\|(\d+)\|FirstId\|(\d+)\|SecondId\|(\d+)\|FirstFailure\|0\|SecondFailure\|0', text())
    assert preemption and int(preemption[1]) == 32 and int(preemption[2]) > 0 and int(preemption[3]) > 0 and preemption[2] != preemption[3], 'Native overlapping catch/GC/preemption regression failed'
    result['NativeThreadPreemption'] = {'Passed': True, 'Checks': int(preemption[1]), 'FirstId': int(preemption[2]), 'SecondId': int(preemption[3]),
        'SingleCpu': True, 'GcStopTheWorld': True, 'Meaning': 'Two real scheduler threads in active catch scopes, IRQ progress, rethrow identity and four collections.'}
    tls = re.search(r'XFE_THREAD_TLS\|Checks\|32\|FirstManagedId\|(\d+)\|SecondManagedId\|(\d+)\|OrdinaryThreadsOnly\|1', text())
    assert tls and int(tls[1]) > 0 and int(tls[2]) > 0 and tls[1] != tls[2], 'Native ThreadStatic/current-thread identity isolation failed'
    result['NativeThreadTls'] = {'Passed': True, 'Checks': 32, 'FirstManagedId': int(tls[1]), 'SecondManagedId': int(tls[2])}
    weak_lifetime = re.search(r'^XFE_WEAK_LIFETIME_RESULT\|True\|Checks\|5\|Weak\|1\|Live\|1\|Cycle\|1'
        r'\|FixedPoint\|1\|Disposed\|1\|HandlesBefore\|(\d+)\|HandlesAfter\|(\d+)\|Failure\|0\r?\n', text(), re.M)
    if weak_lifetime:
        assert weak_lifetime[1] == weak_lifetime[2], 'Native weak-lifetime self-test leaked a GC handle'
        result['NativeWeakLifetime'] = {'Included': True, 'Passed': True, 'Checks': 5,
            'HandlesBefore': int(weak_lifetime[1]), 'HandlesAfter': int(weak_lifetime[2]),
            'Meaning': 'Native weak/strong handles, dead key-value cycle, ephemeron fixed point and disposed table release.'}
    elif prior_receipt and 'XFE_WEAK_LIFETIME_RESULT|' not in text():
        result['NativeWeakLifetime'] = {'Included': False, 'Passed': False, 'NotIncludedHistorical': True,
            'Meaning': 'This immutable historical image predates the weak-lifetime gate; reuse does not qualify it for the new contract.'}
    else:
        raise AssertionError('New native image has no complete passing weak-lifetime/handle-count self-test')
    frozen_log = text()
    frozen_records = re.findall(r'^XFE_FROZEN_RANGE_RESULT\|[^\r\n]*(?:\r?\n|$)', frozen_log, re.M)
    if len(frozen_records) == 1 and re.fullmatch(
            r'XFE_FROZEN_RANGE_RESULT\|True\|Checks\|8\|RegisteredValid\|1\|Failure\|0\r?\n', frozen_records[0]):
        result['NativeFrozenRange'] = {'Included': True, 'Passed': True, 'Checks': 8, 'RegisteredValid': True,
            'Meaning': 'Native startup frozen-object range registration and validation self-test.'}
    elif prior_receipt and 'XFE_FROZEN_RANGE_RESULT|' not in frozen_log:
        result['NativeFrozenRange'] = {'Included': False, 'Passed': False, 'NotIncludedHistorical': True,
            'Meaning': 'This immutable historical image predates the frozen-range gate; reuse does not qualify it for the new contract.'}
    else:
        raise AssertionError('Native image has no single complete passing frozen-range self-test')
    for field, marker, pattern in [
        ('NativeOwnedWeakCollections', 'XFE_OWNED_WEAK_COLLECTIONS_RESULT',
         r'True\|Checks\|6\|NativeCountChecked\|1\|HandlesBefore\|(\d+)\|HandlesAfter\|(\d+)\|Failure\|0'),
        ('NativeJavaScriptWeakCollections', 'XFE_JS_WEAK_COLLECTIONS_RESULT',
         r'True\|NativeCountChecked\|1\|HandlesBefore\|(\d+)\|HandlesAfter\|(\d+)'
         r'\|Created\|(\d+)\|Disposed\|(\d+)\|Tables\|0\|Entries\|0'
         r'\|EngineCachesExercised\|1\|ConstructorFailureReleased\|1\|ConstructorExceptionPreserved\|1'
         r'\|Actual\|true,true,true,true,true,3,true,true,24384,8128,true')]:
        records = re.findall(r'^' + marker + r'\|[^\r\n]*(?:\r?\n|$)', text(), re.M)
        match = re.fullmatch(marker + r'\|' + pattern + r'\r?\n', records[0]) if len(records) == 1 else None
        if match:
            assert match[1] == match[2], field + ' did not restore the native handle count'
            proof = {'Included': True, 'Passed': True, 'NativeCountChecked': True,
                     'HandlesBefore': int(match[1]), 'HandlesAfter': int(match[2])}
            if field == 'NativeOwnedWeakCollections':
                proof['Checks'] = 6
            else:
                assert int(match[3]) > 0 and match[3] == match[4], 'Actual JS weak slots were not released'
                proof.update({'Created': int(match[3]), 'Disposed': int(match[4]), 'TablesAfterRelease': 0,
                              'EntriesAfterRelease': 0, 'ActualInterpreterIntrinsics': True,
                              'EngineCachesExercised': True, 'ConstructorFailureReleased': True,
                              'ConstructorExceptionPreserved': True})
            result[field] = proof
        elif prior_receipt and marker + '|' not in text():
            result[field] = {'Included': False, 'Passed': False, 'NotIncludedHistorical': True}
        else:
            raise AssertionError('Native image has no single complete passing ' + field + ' self-test')
    wait_log('XFE_DESKTOP_APPEARANCE|Dark|657DE2|Persistent|0')
    if 'BootPlatform' in result['PrivateRuntimePatches']:
        topology = re.search(r'XFE_CPU_TOPOLOGY\|Physical\|(\d+)\|Scheduler\|1\|StartedCpus\|1\|BootstrapLogicalCpu\|0', text())
        assert topology and int(topology[1]) == options.virtual_cpus, 'Discovered CPUs and single-CPU execution model differ'
        assert '[LAPIC] Timer tick ' not in text() and '[STRIDE] OnTick:' not in text(), 'Periodic timer diagnostics still run'
        result['CpuTopology'] = {'Passed': True, 'Discovered': int(topology[1]), 'Scheduler': 1, 'Started': 1, 'BootstrapLogical': 0}
        result['PeriodicTimerDiagnosticsAbsent'] = True
    if options.network_device == 'none':
        wait_log('XFE_NETWORK_UNAVAILABLE')
        result['OfflineStartupPassed'] = True
    screenshot('Desktop')
    display_marker = next(line for line in text().splitlines() if line.startswith('XFE_DISPLAY_NATIVE|'))
    display_parts = display_marker.split('|')
    native_width, native_height, scale_percent = int(display_parts[1]), int(display_parts[2]), int(display_parts[4])
    scale = scale_percent / 100
    result['NativeWidth'] = native_width; result['NativeHeight'] = native_height; result['ScalePercent'] = scale_percent
    with Image.open(OUT / 'Desktop.png') as display_image:
        assert display_image.size == (native_width, native_height), 'Screenshot and kernel native geometry differ'
        background_point = (native_width - round(140 * scale), native_height - round(50 * scale))
        background_rgb = display_image.convert('RGB').getpixel(background_point)
        assert background_rgb == (9, 18, 27), ('Fresh desktop did not paint the default dark background', background_rgb)
        gradient_point = (background_point[0], round(24 * scale))
        gradient_rgb = display_image.convert('RGB').getpixel(gradient_point)
        assert all(top > base for top, base in zip(gradient_rgb, background_rgb)), ('The desktop gradient did not reach the actual framebuffer', gradient_rgb, background_rgb)
        work_height = native_height - round(48 * scale)
        brand_bounds = (native_width // 2 - round(150 * scale), work_height // 2 - round(30 * scale),
                        native_width // 2 + round(150 * scale), work_height // 2 + round(40 * scale))
        brand_crop = display_image.convert('RGB').crop(brand_bounds)
        brand_ink = 0
        brand_columns = set()
        for index, (r, g, b) in enumerate(brand_crop.getdata()):
            if r + g + b >= 220:
                brand_ink += 1
                brand_columns.add(index % brand_crop.width)
        assert brand_ink > 100 * scale * scale and len(brand_columns) >= 140 * scale, 'The XFESystem wordmark has no wide actual framebuffer glyph ink'
    result['DefaultAppearance'] = {'Passed': True, 'Mode': 'Dark', 'Accent': '#657DE2',
        'FirstBootWithoutSavedSettings': True, 'BackgroundRgb': list(background_rgb),
        'BackgroundPoint': list(background_point), 'GradientPoint': list(gradient_point), 'GradientRgb': list(gradient_rgb),
        'WordmarkNativeBounds': list(brand_bounds), 'WordmarkInkPixels': brand_ink, 'WordmarkInkColumns': len(brand_columns), 'Screenshot': str(OUT / 'Desktop.png')}
    if options.resolution == '0x0':
        assert (native_width, native_height) == (edid_width, edid_height), 'Automatic GOP mode differs from the virtual monitor native resolution'
    if options.smoke_display:
        result['Checks'] = ['UEFI GPT/FAT32 boot', 'native framebuffer resolution and independent UI scale',
                            'guest runtime variance, browser JS/Markdown and managed PE/IL startup regressions']
        result['ResolutionRequested'] = options.resolution
        result['DisplayMarker'] = next(line for line in text().splitlines() if line.startswith('XFE_DISPLAY_NATIVE|'))
        result['Passed'] = True
        qmp('quit')
        process.wait(timeout=10)
        raise SystemExit(0)
    pointer = [native_width // 2, native_height // 2]
    if options.pointer_cadence:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestPointerCadence import run_pointer_cadence, require_backing
        require_backing(qmp('query-block'), DISK)
        mice = qmp('query-mice')
        result['PointerDevices'] = mice
        current = [mouse for mouse in mice if mouse.get('current')]
        assert len(current) == 1 and not current[0].get('absolute'), 'Expected exactly one active relative pointer'
        if options.usb_input:
            tree = qmp('human-monitor-command', {'command-line': 'info qtree'})
            assert 'dev: i8042' not in tree and 'dev: ps2-mouse' not in tree, 'PS/2 remained enabled'
            assert len(mice) == 1 and tree.count('dev: usb-mouse,') == 1 and 'dev: usb-mouse, id "XfeMouse"' in tree, 'The sole input device is not the owned USB mouse'
            result['UsbMouseTopologyVerified'] = True
        cadence_output = OUT / 'PointerCadence'
        cadence_output.mkdir()
        measured = run_pointer_cadence(qmp, text, screenshot, native_width, native_height, pointer, cadence_output,
            seconds=5, input_rate=1000, target_hz=options.refresh_hint)
        result['PointerCadence'] = measured
        result['PointerDevices'] = mice
        result['PointerCadenceRequirementMet'] = measured['Passed']
        result['PhysicalMonitorRefreshMatched'] = False
        # This receipt qualifies a healthy boot and complete measurement only.
        # The requested cadence has its own explicit, unrelaxed acceptance.
        result['Checks'] = ['Actual boot and framebuffer', 'Bound pointer input/presentation timing evidence',
                            'Physical monitor refresh and GPU/VSync remain unverified']
        result['Passed'] = measured['EvidenceValid']
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    if options.desktop_documents:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestDesktopDocuments import run_desktop_documents
        result['DesktopDocuments'] = run_desktop_documents(qmp, text, wait_log, screenshot, native_width, native_height,
            scale, pointer, desktop_snapshots)
        result['TargetedDocumentWorkflowProfile'] = True
        result['BingCssParityPassed'] = False
        result['Passed'] = result['DesktopDocuments']['Passed']
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    if options.editor_stability:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestEditorStability import run_editor_stability
        from NativeDependentSlotEvidence import NativeDependentSlotReader
        result['TargetedEditorStabilityProfile'] = True
        ownership_reader = NativeDependentSlotReader(qmp, ELF, result['KernelSha256'], OUT / 'NativeOwnership',
            ROOT / 'Source/XFESystem.Kernel/obj/Release/net10.0/linux-x64/cosmos/ref/System.Private.CoreLib.dll',
            result['PrivateRuntimePatches']['GcTail']) if options.stability_cycles else None
        try:
            result['EditorStability'] = run_editor_stability(qmp, text, wait_log, screenshot, native_width, native_height,
                scale, pointer, desktop_snapshots, options.stability_idle_seconds, options.stability_cycles,
                native_reader=ownership_reader)
        finally:
            stability_record = OUT / 'EditorStability.json'
            if stability_record.exists():
                result['EditorStability'] = json.loads(stability_record.read_text(encoding='utf-8'))
        result['Passed'] = result['EditorStability']['Passed']
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    if options.editor_responsiveness:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestEditorResponsiveness import run_editor_responsiveness
        result['EditorResponsiveness'] = run_editor_responsiveness(qmp, text, wait_log, screenshot, native_width, native_height, scale, pointer, desktop_snapshots)
        result['TargetedEditorProfile'] = True
        result['Passed'] = result['EditorResponsiveness']['Passed']
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    if options.document_first_paint:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestDocumentFirstPaint import run_document_first_paint
        result['DocumentFirstPaint'] = run_document_first_paint(qmp, text, wait_log, screenshot, native_width, native_height, scale, pointer)
        result['TargetedPreviewProfile'] = True
        result['FirstPaintRequirementMet'] = result['DocumentFirstPaint']['FirstPaintRequirementMet']
        result['CudaRequirementMet'] = False
        result['Passed'] = True
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    if options.preview_performance:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestPreviewPerformance import run_preview_performance
        result['PreviewPerformance'] = run_preview_performance(qmp, text, wait_log, screenshot, native_width, native_height, scale, pointer)
        result['TargetedPreviewProfile'] = True
        result['Passed'] = True
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    if options.browser_oom_negative:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestBrowserOom import run_browser_oom_check
        result['Kind'] = 'NativeOomNegativeProof'
        result['NativeOomNegative'] = run_browser_oom_check(qmp, text, wait_log, scale, pointer)
        result['Passed'] = False  # A deliberately halted guest cannot qualify for packaging or launch.
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    if options.browser_pipeline:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestBrowserPipeline import run_browser_checks
        browser_checks = run_browser_checks(qmp, text, wait_log, screenshot, native_width, native_height, scale, pointer)
        result['BrowserPipeline'] = browser_checks
        result['Checks'] = ['UEFI GPT/FAT32 boot', 'native framebuffer and managed startup regressions'] + browser_checks['Checks']
        result['Passed'] = True
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    if options.desktop_ui_only:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestDesktopUi import run_desktop_checks
        desktop_ui = run_desktop_checks(qmp, text, wait_log, screenshot, native_width, native_height, scale, pointer, options.input_rate, options.refresh_hint, options.desktop_v4)
        result['Checks'] = ['UEFI GPT/FAT32 boot', 'native framebuffer geometry and independent UI scale', 'guest startup runtime, managed IL and JS/Markdown regressions']
        result['DesktopUiChecks'] = desktop_ui['Checks']; result['DesktopPerformance'] = desktop_ui['Performance']
        result['DesktopVisualEvidence'] = desktop_ui.get('VisualEvidence', {})
        result['DesktopUiPassed'] = True; result['TargetedDesktopProfile'] = True; result['NativeHttpPassed'] = False
        result['Passed'] = True
        qmp('quit'); process.wait(timeout=10)
        raise SystemExit(0)
    def click(x, y):
        target = [round(x * scale), round(y * scale)]
        while pointer != target:
            dx = max(-80, min(80, target[0] - pointer[0]))
            dy = max(-80, min(80, target[1] - pointer[1]))
            qmp('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)
        qmp('input-send-event', {'events': [{'type': 'btn', 'data': {'down': True, 'button': 'left'}}]})
        time.sleep(.06)
        qmp('input-send-event', {'events': [{'type': 'btn', 'data': {'down': False, 'button': 'left'}}]})
        time.sleep(.1)
    def key(chord):
        qmp('human-monitor-command', {'command-line': f'sendkey {chord}'})
        time.sleep(.15)
    def wait_desktop_state(active_id, expected_windows, start, description, timeout=90):
        deadline = time.monotonic() + timeout
        last_state = None
        while time.monotonic() < deadline:
            log = text()
            states = desktop_snapshots(log[start:])
            if states:
                last_state = states[-1]
                actual_windows = {window_id: state['Application'] for window_id, state in last_state['Windows'].items()}
                if (last_state['ActiveId'] == active_id and actual_windows == expected_windows and
                        last_state['StartMenuOpen'] == 0 and last_state['WindowSwitching'] == 0 and
                        all(not state['Minimized'] for state in last_state['Windows'].values())):
                    acknowledgement = dict(last_state, Action=description, SerialOffset=start + last_state['SerialOffset'])
                    result.setdefault('DesktopTransitionAcknowledgements', []).append(acknowledgement)
                    return acknowledgement
            if any(marker in log for marker in ('CPU EXCEPTION:', '[FAILFAST]', '*** UNHANDLED EXCEPTION')):
                raise RuntimeError('Guest exception while waiting for ' + description + '\n' + log[-4000:])
            if process.poll() is not None:
                raise RuntimeError('Guest exited while waiting for ' + description)
            time.sleep(.05)
        raise AssertionError((description, 'No complete matching native desktop acknowledgement',
                              {'ActiveId': active_id, 'Windows': expected_windows}, last_state, text()[start:][-5000:]))
    click(60, 120)
    wait_log('XFE_DESKTOP_ACTIVE|Files')
    screenshot('Files')
    click(370, 116)
    wait_log('XFE_DESKTOP_ACTIVE|Editor')
    for character in 'hello': key(character)
    first_save = len(text())
    key('ctrl-s')
    wait_log('XFE_DESKTOP_SAVED|X:\\Users\\Default\\Documents\\Document1.txt|5', first_save)
    screenshot('Editor-Saved')
    key('esc')
    wait_log('XFE_DESKTOP_ACTIVE|Files', first_save)
    click(220, 170)
    reload_start = len(text())
    key('ctrl-s')
    wait_log('XFE_DESKTOP_SAVED|X:\\Users\\Default\\Documents\\Document1.txt|5', reload_start)
    reload_close_start = len(text())
    key('esc')
    wait_desktop_state(1, {1: 'Files'}, reload_close_start, 'Close reloaded text editor before opening JavaScript')
    script_start = len(text())
    click(220, 222)
    # Local .js files now open the real Monaco model; F5 is a website-style
    # source preview, while browser scripts retain their independent execution
    # tests below. Do not count the old console-output marker as this proof.
    import re
    model_pattern = re.compile(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END', re.S)
    expected_script = "const greeting = 'XFESystem';\nconsole.log(`Hello from ${greeting}!`);\n"
    model_deadline = time.monotonic() + 90
    actual_model = ''
    while time.monotonic() < model_deadline:
        models = list(model_pattern.finditer(text()[script_start:]))
        if models:
            actual_model = models[-1].group(1).replace('\r', '')
            if actual_model == expected_script: break
        assert all(marker not in text() for marker in ('CPU EXCEPTION:', '[FAILFAST]', '*** UNHANDLED EXCEPTION')), text()[-4000:]
        time.sleep(.1)
    assert actual_model == expected_script, ('Desktop JavaScript file was not loaded into the actual Monaco model', actual_model)
    script_saved = len(text());key('ctrl-s')
    wait_log('XFE_DESKTOP_SAVED|X:\\Users\\Default\\Documents\\Hello.js|' + str(len(expected_script)), script_saved, 30)
    screenshot('Editor-Monaco-JavaScript')
    key('f5')
    sys.path.insert(0, str(ROOT / 'Tools'))
    from TestWebsiteDocument import wait_document, verify_code, capture_document, run_fixture_checks
    script_proof = wait_document(text,script_start,lambda proof:any(block.get('CodeText')==expected_script for block in proof.get('CodeBlocks',[])))
    verify_code(script_proof,'javascript',expected_script)
    script_frame = capture_document(screenshot,'Editor-JavaScript-Website-Preview',text,script_proof,scale)
    result['EditorJavaScriptWebsitePreview'] = {'MonacoOriginalModel': actual_model,'CodeDomReadback':script_proof,'Framebuffer':script_frame,'Passed':True,
        'Behavior':'F5 previews source with website highlighting; source is not executed as a local program.'}
    screenshot('Editor-JavaScript-Preview')
    preview_close_start = len(text())
    key('esc')
    wait_desktop_state(4, {1: 'Files', 4: 'Editor'}, preview_close_start, 'Close JavaScript preview and return to its Monaco editor')
    editor_close_start = len(text())
    key('alt-f4')  # Monaco owns Esc; close the saved editor with the real desktop shortcut.
    wait_desktop_state(1, {1: 'Files'}, editor_close_start, 'Close JavaScript Monaco editor before opening Readme')
    markdown_start = len(text())
    click(220, 250)
    wait_desktop_state(6, {1: 'Files', 6: 'Editor'}, markdown_start, 'Open Readme text editor before F5')
    key('f5')
    wait_log('XFE_EDITOR_MARKDOWN_PREVIEW_OK', markdown_start)
    wait_desktop_state(7, {1: 'Files', 6: 'Editor', 7: 'Browser'}, markdown_start, 'Readme F5 opens its own web preview')
    markdown_proof = wait_document(text,markdown_start,lambda proof:(proof.get('Heading') or {}).get('Text')=='XFESystem Markdown')
    assert 'Website renderer' in markdown_proof['BodyText']
    markdown_frame = capture_document(screenshot,'Editor-Markdown-Website-Preview',text,markdown_proof,scale,require_heading=True)
    result['EditorMarkdownWebsitePreview'] = {'DomReadback':markdown_proof,'Framebuffer':markdown_frame,'Passed':True}
    screenshot('Editor-Markdown-Preview')
    result['WebsiteDocument'] = run_fixture_checks(qmp,text,screenshot,scale,pointer)
    fixture_close_start = len(text())
    key('alt-f4')  # The fixture toolbar click focused web content, which owns Esc.
    wait_desktop_state(6, {1: 'Files', 6: 'Editor'}, fixture_close_start, 'Close Markdown fixture browser and return to Readme')
    markdown_close_start = len(text())
    key('esc')
    wait_desktop_state(1, {1: 'Files'}, markdown_close_start, 'Close Readme editor before navigating to the managed program')
    for _ in range(3): click(180, 116)
    click(220, 170)
    click(220, 170)
    application_start = len(text())
    click(220, 170)
    wait_log('XFE_MANAGED_APP_OK', application_start)
    wait_log('XFE_MANAGED_XFECONSOLE_OK', application_start)
    wait_log('XFE_DESKTOP_ACTIVE|Terminal', application_start)
    screenshot('Terminal-ManagedAssembly')
    key('esc'); key('esc')
    browser_start = len(text())
    click(60, 306)
    wait_log('XFE_BROWSER_START_DOM_RENDER_OK', browser_start)
    screenshot('Browser-Start-JavaScript')
    fixture_nonce = 'Guest-' + uuid.uuid4().hex
    fixture = subprocess.Popen([sys.executable,
        str(ROOT / 'Tests/XFESystem.HttpProtocol.Tests/Fixtures/NativeHttpServer.py'),
        '--Port', '0', '--Nonce', fixture_nonce, '--Log', str(OUT / 'HttpPeer.jsonl')],
        cwd=ROOT, stdout=subprocess.PIPE, stderr=(OUT / 'HttpPeerError.log').open('wb'), text=True)
    peer = json.loads(fixture.stdout.readline())
    (OUT / 'HttpPeer.json').write_text(json.dumps(peer, ensure_ascii=False, indent=2), encoding='utf-8')
    def type_text(value):
        names = {':': 'shift-semicolon', '/': 'slash', '.': 'dot', '-': 'minus'}
        for character in value:
            key(names.get(character, 'shift-' + character.lower() if character.isupper() else character))
    def navigate(path, scheme='http', timeout=15):
        start = len(text())
        address = f'{scheme}://10.0.2.2:{peer["Port"]}{path}'
        key('ctrl-l')
        type_text(address)
        key('ret')
        wait_log('XFE_BROWSER_TEXT_END', start, timeout)
        log = text()[start:]
        expected_address = address[:-len(path)] + '/Index' if path == '/Redirect' else address
        assert 'XFE_BROWSER_NAVIGATED|' + expected_address in log, log
        body = log.split('XFE_BROWSER_TEXT_BEGIN\n', 1)[1].split('\nXFE_BROWSER_TEXT_END', 1)[0]
        return body, log
    network_checks = []
    wait_log('XFE_NETWORK_DHCP_OK|10.0.2.15|DnsServers|1')
    for endpoint in ['/Index', '/Chunked', '/Redirect']:
        body, navigation_log = navigate(endpoint)
        assert fixture_nonce in body and 'Native network JS: 42' in body, body
        assert 'XFE_BROWSER_UTF8_BODY_OK' in navigation_log, navigation_log
        network_checks.append('native TCP/HTTP+UTF8+DOM JavaScript ' + endpoint)
        screenshot('Browser-HTTP-' + endpoint[1:])
    body, _ = navigate('/Csp')
    assert 'CSP original' in body and 'Automatic scripts blocked' in body and 'CSP should block this' not in body, body
    network_checks.append('native HTTP CSP blocks automatic script')
    body, _ = navigate('/Index', scheme='https')
    assert 'HTTPS is unavailable' in body and 'Certificate validation cannot be bypassed' in body, body
    network_checks.append('native HTTPS explicitly refused without a verified TLS port')
    body, _ = navigate('/Index')
    assert fixture_nonce in body and 'Native network JS: 42' in body, body
    network_checks.append('successful native navigation after HTTPS refusal releases browser gate')
    body, _ = navigate('/Truncated')
    assert 'Navigation failed' in body and 'declared length' in body, body
    network_checks.append('native HTTP truncated response rejected')
    body, _ = navigate('/Index')
    assert fixture_nonce in body and 'Native network JS: 42' in body, body
    network_checks.append('successful native navigation after truncated response releases socket and browser gate')
    body, _ = navigate('/Slow', timeout=25)
    assert 'Navigation failed' in body and ('timed out' in body or 'transport stopped' in body), body
    network_checks.append('native HTTP deadline or transport-fault refusal preserves GUI')
    resumed = len(text())
    key('ctrl-l'); type_text('about:Start'); key('ret')
    wait_log('XFE_BROWSER_TEXT_END', resumed)
    assert 'JavaScript ready: 12' in text()[resumed:], text()[resumed:]
    network_checks.append('offline page remains usable after native HTTP failure')
    peer_requests = [json.loads(line) for line in (OUT / 'HttpPeer.jsonl').read_text(encoding='utf-8').splitlines()]
    assert all(any(request['Path'] == endpoint for request in peer_requests) for endpoint in
               ['/Index', '/Chunked', '/Redirect', '/Csp', '/Truncated']), peer_requests
    result['NativeHttpChecks'] = network_checks
    result['NativeHttpPassed'] = True
    result['HttpPeerRequests'] = peer_requests
    result['HttpPeerNonce'] = fixture_nonce
    if options.desktop_ui:
        sys.path.insert(0, str(ROOT / 'Tools'))
        from TestDesktopUi import run_desktop_checks
        desktop_ui = run_desktop_checks(qmp, text, wait_log, screenshot, native_width, native_height, scale, pointer, options.input_rate, options.refresh_hint, options.desktop_v4)
        result['DesktopUiChecks'] = desktop_ui['Checks']
        result['DesktopPerformance'] = desktop_ui['Performance']
        result['DesktopVisualEvidence'] = desktop_ui.get('VisualEvidence', {})
        result['DesktopUiPassed'] = True
    result['Checks'] = ['UEFI GPT/FAT32 boot', 'NativeAOT C# kernel',
                        'real XFEConsole 2.6.0 API execution in guest',
                        'guest ANSI RGB model', 'guest Console color PAL',
                        'mouse opens Files and Editor', 'PS2 keyboard edits and Ctrl+S saves',
                        'file reopened from RAM live storage with preserved content length',
                        'editor hosts the real Monaco model and previews JavaScript source through website highlighting',
                        'editor previews Markdown through website Markdig pipeline',
                        'mouse launches independent managed PE/IL file from Programs',
                        'managed IL interpreter executes Console, ANSI and real XFEConsole bridge',
                        'browser about:Start executes inline DOM JavaScript and displays result']
    result['GuestStatus'] = qmp('query-status')
    result['Passed'] = True
    qmp('quit')
    process.wait(timeout=10)
except Exception as error:
    result['Passed'] = False
    result['Failure'] = str(error)
    # Inspect only this private guest's bounded page-allocation table. This
    # distinguishes total-memory exhaustion from contiguous-run fragmentation
    # before finally closes QEMU; it never changes guest memory or acceptance.
    try:
        failure_log = text()
        rat = re.search(r'RAT location: 0x([0-9A-Fa-f]+)', failure_log)
        offset = re.search(r'HHDM offset: (?:0x)+([0-9A-Fa-f]+)', failure_log)
        pages = re.search(r'\[PageAllocator\] Total pages: (\d+),', failure_log)
        if client is not None and 'qmp' in globals() and rat and offset and pages:
            physical = int(rat[1], 16) - int(offset[1], 16)
            count = int(pages[1])
            assert 0 < count <= 1_048_576 and 0 <= physical < options.memory_mib * 1024 * 1024
            assert physical + count <= options.memory_mib * 1024 * 1024
            table_path = OUT / 'FailurePageAllocationTable.bin'
            quoted_path = table_path.as_posix()
            assert not any(character in quoted_path for character in ('"', '\n', '\r'))
            qmp('human-monitor-command', {'command-line': f'pmemsave {physical:#x} {count} "{quoted_path}"'})
            table = table_path.read_bytes()
            assert len(table) == count
            longest = current = free_pages = 0
            for value in table:
                if value == 0:
                    current += 1; free_pages += 1; longest = max(longest, current)
                else:
                    current = 0
            result['FailurePageAllocationTable'] = {'ReadOnly': True, 'Path': str(table_path),
                'Sha256': hashlib.sha256(table).hexdigest(), 'PhysicalAddress': physical, 'Entries': count,
                'FreePages': free_pages, 'LargestContiguousFreePages': longest,
                'Meaning': 'One byte per guest page, PageType.Empty=0; diagnostic snapshot does not repair or qualify the image.'}
    except Exception as memory_diagnostic_error:
        result['MemoryDiagnosticFailure'] = str(memory_diagnostic_error)
    # Preserve register and frame-pointer evidence for a hang before GUI-ready.
    # Diagnostics never turn an incomplete boot into a successful result.
    try:
        if client is None:
            client = socket.create_connection(('127.0.0.1', port), timeout=2)
            stream = client.makefile('rwb')
            stream.readline()
            stream.write(b'{"execute":"qmp_capabilities"}\n'); stream.flush()
            while True:
                response = json.loads(stream.readline())
                if 'return' in response: break
                if 'error' in response: raise RuntimeError(response['error'])
        def diagnostic_hmp(command):
            if 'qmp' in globals():
                return qmp('human-monitor-command', {'command-line': command})
            stream.write((json.dumps({'execute': 'human-monitor-command',
                'arguments': {'command-line': command}}) + '\n').encode())
            stream.flush()
            while True:
                response = json.loads(stream.readline())
                if 'return' in response: return response['return']
                if 'error' in response: raise RuntimeError(response['error'])
        registers = diagnostic_hmp('info registers')
        (OUT / 'FailureRegisters.json').write_text(json.dumps({'return': registers, 'ReadOnly': True}), encoding='utf-8')
        stack = bounded_failure_stack(registers, diagnostic_hmp)
        dump_path = OUT / 'FailureStack.txt'
        dump_path.write_text(stack.pop('RawDump'), encoding='utf-8')
        stack.update({'Path': str(dump_path), 'Sha256': hashlib.sha256(dump_path.read_bytes()).hexdigest()})
        (OUT / 'FailureStack.json').write_text(json.dumps(stack, indent=2), encoding='utf-8')
        result['FailureStack'] = stack
    except Exception as diagnostic_error:
        result['DiagnosticFailure'] = str(diagnostic_error)
    raise
finally:
    if client:
        client.close()
    if process.poll() is None:
        process.terminate()
        process.wait(timeout=10)
    if fixture is not None and fixture.poll() is None:
        fixture.terminate()
        fixture.wait(timeout=5)
    result['QemuExitCode'] = process.returncode
    result['ImageSha256'] = image_hash
    result['ImageVerification'] = image_verification
    result['ImageUnchanged'] = hashlib.sha256(DISK.read_bytes()).hexdigest() == image_hash
    result['DisplayMarker'] = next((line for line in text().splitlines() if line.startswith('XFE_DISPLAY_NATIVE|')), None)
    guest_log = text() if (OUT / 'Serial.log').exists() else ''
    result['NativeGpuDiagnostics'] = [line for line in guest_log.splitlines() if line.startswith('XFE_NVIDIA_NATIVE_')]
    result['CudaExecutionVerified'] = False
    result['GpuRasterizationVerified'] = False
    result['ManagedIlSucceeded'] = all(marker in guest_log for marker in ['XFE_MANAGED_APP_OK', 'XFE_MANAGED_XFECONSOLE_OK', 'XFE_MANAGED_ANSI_RGB_OK'])
    result['CompletedGuestMarkers'] = [line for line in guest_log.splitlines() if line.startswith('XFE_')]
    (OUT / 'Result.json').write_text(json.dumps(result, indent=2, ensure_ascii=False), encoding='utf-8')
    print(OUT)
    print(json.dumps(result, ensure_ascii=False))