sources/qualification-snapshot/scripts/native_candidate_prefix.py

"""Independent bounded prefix-format and host identity validation for candidates.

Only native ARM macOS prefix runtime identity is implemented here. Other hosts
are refused explicitly. The native importer additionally checks packed layout
and every finite BF16 value against its actual loaded model before publication.
"""
import hashlib
from pathlib import Path
import platform
import struct
import subprocess

HEADER_BYTES = 268
PAYLOAD_BYTES = 18452480
ARTIFACT_BYTES = HEADER_BYTES + 68*24 + PAYLOAD_BYTES + 32


def require(condition, message):
    if not condition:
        raise ValueError(message)


def sha(path):
    h=hashlib.sha256()
    with Path(path).open('rb') as f:
        for block in iter(lambda:f.read(1024*1024),b''):h.update(block)
    return h.hexdigest()


def runtime_identity(binary):
    """Reproduce stream_common.rs bytes using independent host queries.

    Candidate CLI is fixed to A8, four workers, user-initiated QoS and a private
    pool driver. The binary hash binds compiled features; Mach-O verifies ARCH.
    """
    require(platform.system()=='Darwin' and platform.machine()=='arm64',
            'independent prefix runtime identity currently requires native ARM macOS')
    with Path(binary['path']).open('rb') as f:
        magic,cpu=struct.unpack('<II',f.read(8))
    require(magic==0xfeedfacf and cpu==0x0100000c,'prefix binary is not a native ARM64 Mach-O')
    require(sha(binary['path'])==binary['sha256'],'prefix binary identity changed')
    def query(args):return subprocess.check_output(args,timeout=10)
    dot=query(['sysctl','-n','hw.optional.arm.FEAT_DotProd'])
    require(dot.strip() in (b'0',b'1'),'unknown host DotProd feature state')
    backend='neon_dotprod' if dot.strip()==b'1' else 'portable_int8'
    uname=query(['uname','-s','-r','-m'])
    brand=query(['sysctl','-n','machdep.cpu.brand_string'])
    build=query(['sw_vers','-buildVersion'])
    data=(bytes.fromhex(binary['sha256'])+b'aarch64macos'+backend.encode()
          +b'\0worker_qos=user-initiated\0pool_driver=\x01'+uname+brand+build)
    return dict(sha256=hashlib.sha256(data).hexdigest(),backend=backend,
                worker_qos='user-initiated',pool_driver=True,
                uname_stdout=uname.decode(),cpu_stdout=brand.decode(),os_build_stdout=build.decode(),
                method='independent reconstruction of stream_common::runtime_identity byte contract')


def inspect_prefix(path, language, assets, profile):
    """Validate requested startup/delay plus every header, record and hash."""
    require(language in ('en','zh'),'unsupported prefix language')
    require(profile['gear']==4 and profile['first_batch']==1 and profile['mode'] in ('hf','realtime')
            and profile['delay'] in (1,2,3,6),'unsupported prefix execution profile')
    startup=18 if profile['mode']=='realtime' else 19+profile['delay']
    require(profile['startup']==startup,'requested startup/prefill mismatch')
    magic=b'A8PFX003' if profile['mode']=='realtime' or profile['delay']<3 else (
        b'A8PFX002' if profile['delay']==3 else b'A8PFX001')
    path=Path(path).resolve()
    require(path.stat().st_size==ARTIFACT_BYTES,'prefix artifact length differs')
    runtime=runtime_identity(assets['runtime_binary'])
    with path.open('rb') as f:
        header=f.read(HEADER_BYTES)
        require(len(header)==HEADER_BYTES and header[:8]==magic,'prefix format/startup version differs')
        values=struct.unpack('<25I',header[8:108])
        expected=(dict(en=151668,zh=151667)[language],4,profile['delay'],18,16000,400,160,840,40,
                  68,17,32,36,2048,256,16,1,4,1,1,startup,128,151644,151665,151645)
        require(values==expected,'prefix language/delay/startup/execution shape differs')
        hashes=[header[i:i+32].hex() for i in range(108,HEADER_BYTES,32)]
        require(all(h!='0'*64 for h in hashes),'prefix empty identity hash')
        require(hashes[1]==assets['frontend']['sha256'] and hashes[2]==assets['runtime_binary']['sha256'],
                'prefix frontend/binary identity differs')
        require(hashes[3]==runtime['sha256'],'prefix host/runtime identity differs')
        with Path(assets['model_stream']['path']).open('rb') as model:
            model.seek(-32,2)
            require(model.read(32).hex()==hashes[0],'prefix model body identity differs')
        digest=hashlib.sha256(header)
        payload=0
        for kind,count,frames,width in ((0,32,68,2048),(1,36,17,256)):
            for index in range(count):
                record=f.read(24)
                require(record==struct.pack('<6I',kind,index,frames,width,0,frames),
                        'prefix per-layer cache shape/position differs')
                digest.update(record)
                remaining=frames*width*2*2
                payload+=remaining
                while remaining:
                    block=f.read(min(65536,remaining))
                    require(bool(block),'truncated prefix KV payload')
                    digest.update(block);remaining-=len(block)
        require(payload==PAYLOAD_BYTES and f.read(32)==digest.digest() and not f.read(1),
                'prefix footer/trailing bytes differ')
    metadata=dict(path=str(path),body_sha256=digest.hexdigest(),payload_bytes=PAYLOAD_BYTES,
        artifact_bytes=ARTIFACT_BYTES,encoder_frames=68,decoder_tokens=17,format=magic.decode(),
        transcription_delay_ms=profile['delay']*80,startup_policy=profile['startup_policy'],
        startup_input_tokens=startup,first_batch_windows=1,model_body_sha256=hashes[0],
        frontend_sha256=hashes[1],math_sha256=hashes[2],runtime_sha256=hashes[3],layout_sha256=hashes[4],
        independent_runtime_validation=runtime,
        remaining_native_validation='Runtime additionally verifies actual packed layout identity and finite BF16 payloads before importing caches.')
    return dict(path=str(path),bytes=ARTIFACT_BYTES,sha256=sha(path)),metadata