XFESystem

"""Validate real AP startup/rendezvous evidence; never infer managed SMP or scanout."""
import re


def validate_multiprocessors(serial, virtual_cpus):
    expected = min(virtual_cpus, 32)
    summary = re.findall(r'^XFE_MULTIPROCESSOR\|Reported\|(\d+)\|Online\|(\d+)\|Backend\|LimineApKernelJobs\|ManagedSchedulerCpus\|1\r?$', serial, re.M)
    assert len(summary) == 1, 'Exactly one complete native MP summary required'
    reported, online = map(int, summary[0])
    assert reported == virtual_cpus and online == expected, 'Configured CPUs did not actually come online'
    entries = re.findall(r'^XFE_AP_ONLINE\|Index\|(\d+)\|ApicId\|(\d+)\|CoreKey\|(\d+)\|PrivateDescriptors\|1\r?$', serial, re.M)
    entries = [tuple(map(int, item)) for item in entries]
    assert len(entries) == expected - 1, 'Missing or duplicate AP startup evidence'
    assert [item[0] for item in entries] == list(range(1, expected)), 'AP indices are missing, repeated or unordered'
    apic_ids = [item[1] for item in entries]
    bsp = re.findall(r'^XFE_CPU_OK bsp_apic=(\d+) cr3=0x[0-9A-Fa-f]+\r?$', serial, re.M)
    assert len(bsp) == 1 and len(set(apic_ids + [int(bsp[0])])) == expected, 'APIC IDs must identify distinct executing CPUs'
    proof = re.findall(r'^XFE_AP_JOBS_PROOF\|Online\|(\d+)\|Rendezvous\|(\d+)\|CopyCases\|32\|Passed\|1\|ManagedSmp\|0\|Physical240Hz\|0\r?$', serial, re.M)
    assert len(proof) == 1 and tuple(map(int, proof[0])) == (expected, expected - 1), 'Actual AP rendezvous/copy test incomplete'
    assert 'XFE_NATIVE_SELFTEST_OK' in serial and not any(marker in serial for marker in (
        'XFE_NATIVE_FAIL', 'XFE_NATIVE_EXCEPTION', 'XFE_RUNTIME_FATAL', 'XFE_ZERO_FAILFAST')), 'Kernel failed during native MP proof'
    assert serial.index('XFE_AP_JOBS_PROOF|') < serial.index('XFE_NATIVE_SELFTEST_OK'), 'AP proof must precede kernel readiness'
    return {'Kind': 'XfeNativeMultiprocessorEvidence1', 'Passed': True,
            'ReportedCpus': reported, 'ExecutingCpus': online, 'ManagedSchedulerCpus': 1,
            'ManagedSmpVerified': False, 'Physical240HzVerified': False,
            'Backend': 'LimineApKernelJobsV1', 'RendezvousParticipants': expected - 1,
            'CopyCases': 32, 'ApplicationProcessors': [dict(Index=i, ApicId=a, CoreKey=c) for i, a, c in entries]}