XFESystem

"""Source-bound guest Bing acceptance. This module performs no host HTTP requests."""
import hashlib
import json
import math
from pathlib import Path
import re
import time
from urllib.parse import parse_qs, urljoin, urlsplit

from PIL import Image

RESPONSE = 'XFE_NATIVE_WEB_RESPONSE'
REMOTE = 'XFE_REMOTE_PAGE_PROOF'
HOME = 'https://www.bing.com/'
NAT_ARGUMENT = 'user,id=NativeNet,net=10.0.2.0/24,host=10.0.2.2,dns=10.0.2.3,dhcpstart=10.0.2.15'
NIC_ARGUMENT = 'e1000,netdev=NativeNet'
TITLE = re.compile(r'bing|必应', re.I)
PRESENTED = re.compile(r'^XFE_BROWSER_BACKGROUND_PRESENTED\|Window\|(\d+)\|Generation\|(\d+)\|Address\|([^|\r\n]+)'
                       r'\|SnapshotSequence\|(\d+)\|PresentationSequence\|(\d+)\|PresentationTicks\|(\d+)\|ScrollY\|([\d.]+)\r?$', re.M)
WINDOW = re.compile(r'^XFE_WINDOW_STATE\|(\d+)\|(\w+)\|(-?\d+)\|(-?\d+)\|(\d+)\|(\d+)\|([01])\|([01])\|([01])\r?$', re.M)
BROWSER_STATE = re.compile(r'XFE_BROWSER_STATE_BEGIN\r?\n(\d+)\|([^\r\n]+)\r?\n(.*?)\r?\nXFE_BROWSER_STATE_END', re.S)


def native_network_profile(enabled):
    """Describe the actual QEMU profile, without claiming guest network success."""
    return {'Kind': 'XfeIndependentNativeNetworkProfile', 'Enabled': bool(enabled),
            'Backend': 'QEMU user NAT' if enabled else 'None', 'Device': 'e1000' if enabled else None,
            'PciVendorDevice': '8086:100e' if enabled else None,
            'GuestAddress': '10.0.2.15' if enabled else None, 'Gateway': '10.0.2.2' if enabled else None,
            'DnsServer': '10.0.2.3' if enabled else None, 'AddressConfiguration': 'Static' if enabled else None,
            'NetdevArgument': NAT_ARGUMENT if enabled else None, 'DeviceArgument': NIC_ARGUMENT if enabled else None,
            'HostProxy': False, 'HostForwarding': False, 'Rtc': 'base=utc', 'Cpu': 'max',
            'EntropyPolicy': 'Guest CPUID plus bounded RDSEED/RDRAND with carry check; unavailable fails closed'}


def network_arguments(enabled):
    return ['-netdev', NAT_ARGUMENT, '-device', NIC_ARGUMENT] if enabled else ['-net', 'none']


def validate_network_arguments(arguments, enabled):
    expected = network_arguments(enabled)
    options = ('-net', '-netdev', '-nic', '-device')
    actual = []
    for index, argument in enumerate(arguments):
        if argument not in options:
            continue
        assert index + 1 < len(arguments), 'Missing QEMU network argument value'
        value = arguments[index + 1]
        actual.extend((argument, value))
    assert actual == expected, 'QEMU network arguments differ from the source-bound profile'
    for key, value in (('-cpu', 'max'), ('-rtc', 'base=utc')):
        indices = [index for index, item in enumerate(arguments) if item == key]
        assert len(indices) == 1 and indices[0] + 1 < len(arguments) and arguments[indices[0] + 1] == value, 'Missing or ambiguous CPU/UTC profile'
    return native_network_profile(enabled)


def window_geometry(log, application):
    end = log.rfind('XFE_DESKTOP_STATE_END')
    begin = log.rfind('XFE_DESKTOP_ACTIVE|', 0, end)
    assert begin >= 0 and end > begin, 'No complete current native desktop state'
    matches = [item for item in WINDOW.finditer(log[begin:end]) if item[2] == application and item[7] == '0']
    assert matches, 'No visible actual ' + application + ' window geometry'
    match = matches[-1]
    return int(match[1]), [int(match[index]) for index in (3, 4, 5, 6)]


def json_blocks(log, marker, after=0):
    expression = re.compile(re.escape(marker) + r'_BEGIN\r?\n(.*?)\r?\n' + re.escape(marker) + r'_END', re.S)
    result = []
    for match in expression.finditer(log, after):
        try:
            value = json.loads(match[1])
        except (ValueError, TypeError):
            continue  # An incomplete/rejected packet never counts as evidence.
        if isinstance(value, dict):
            result.append({'Value': value, 'SerialOffset': match.start(), 'EndOffset': match.end(), 'Raw': match[0]})
    return result


def _url(value):
    parsed = urlsplit(value)
    assert parsed.scheme == 'https' and parsed.hostname and not parsed.username and not parsed.password, 'A Bing hop must use HTTPS without credentials'
    assert parsed.port in (None, 443), 'Unexpected HTTPS port'
    return parsed


def _same_url(left, right):
    try:
        a, b = _url(left), _url(right)
    except (AssertionError, TypeError, ValueError, AttributeError):
        return False
    return (a.hostname.lower(), a.path or '/', a.query) == (b.hostname.lower(), b.path or '/', b.query)


def validate_response_chain(packets, initial=HOME, title_pattern=TITLE):
    """Validate actual guest response packets; redirects cannot become implicit success."""
    expected, chain, seen = initial, [], set()
    for packet in packets:
        value = packet['Value']
        if not _same_url(value.get('RequestUri', ''), expected):
            continue
        uri = _url(value['RequestUri'])
        assert value.get('NativeTransport') is True, 'A host transport cannot qualify'
        assert value.get('CertificateValidated') is True, 'Every HTTPS hop requires validated TLS'
        assert _same_url(value['ResponseUri'], value['RequestUri']), 'The hop response address differs from its request'
        request, headers = value.get('RequestHeaders', ''), value.get('ResponseHeaders', '')
        target = (uri.path or '/') + ('?' + uri.query if uri.query else '')
        assert request.splitlines() and request.splitlines()[0] == 'GET ' + target + ' HTTP/1.1', 'Missing actual HTTP request line'
        host = re.search(r'^Host:\s*([^\r\n]+)', request, re.M | re.I)
        assert host and host[1].strip().lower() in (uri.hostname.lower(), uri.hostname.lower() + ':443'), 'HTTP Host does not match TLS URL'
        status = value.get('StatusCode')
        assert type(status) is int and re.match(r'^HTTP/1\.[01]\s+' + str(status) + r'(?:\s|\r?$)', headers), 'Status differs from raw response headers'
        assert value.get('RequestId') is not None, 'Missing guest request identity'
        chain.append(packet)
        if status in (301, 302, 303, 307, 308):
            location = value.get('Location')
            assert isinstance(location, str) and location, 'Redirect has no Location'
            raw_location = re.search(r'^Location:\s*([^\r\n]+)', headers, re.M | re.I)
            assert raw_location and raw_location[1].strip() == location, 'Redirect differs from raw Location'
            expected = urljoin(value['RequestUri'], location)
            _url(expected)
            assert expected not in seen and len(chain) <= 5, 'Redirect cycle or bound exceeded'
            seen.add(expected)
            continue
        assert status == 200, 'Bing did not return HTTP 200'
        assert uri.hostname.lower() == 'bing.com' or uri.hostname.lower().endswith('.bing.com'), 'The final page is outside Bing'
        assert value.get('ContentType', '').split(';')[0].strip().lower() in ('text/html', 'application/xhtml+xml'), 'The response is not HTML'
        assert value.get('HasHtml') is True and value.get('HasHead') is True, 'No actual HTML/head was received'
        assert title_pattern.search(value.get('HtmlTitle', '')), 'The real HTML title does not match the requested page'
        assert type(value.get('BodyBytes')) is int and value['BodyBytes'] >= 128, 'No bounded actual HTML body'
        assert re.fullmatch(r'[0-9a-fA-F]{64}', value.get('BodySha256', '')), 'Missing actual HTML fingerprint'
        return chain
    raise AssertionError('No complete validated native HTTPS Bing response chain')


def _rect(value):
    values = [float(value[name]) for name in ('X', 'Y', 'Width', 'Height')]
    assert all(math.isfinite(number) for number in values) and values[2] > 0 and values[3] > 0, 'Invalid actual element geometry'
    return values


def validate_remote_page(page, response, title_pattern=TITLE):
    assert _same_url(page['Address'], response['ResponseUri']), 'Remote DOM belongs to another address'
    assert page.get('BodySha256', '').lower() == response['BodySha256'].lower(), 'Rendered HTML does not match the HTTP body'
    assert page.get('CertificateValidated') is True, 'The actual rendered response must retain validated TLS'
    assert type(page.get('SnapshotSequence')) is int and page['SnapshotSequence'] > 0, 'Missing actual immutable paint snapshot identity'
    assert title_pattern.search(page.get('Title', '')) and not re.search(r'failed|unavailable|\berror\b|错误|不可用', page.get('Title', ''), re.I), 'A failure page cannot qualify'
    width, height = float(page['ViewportWidth']), float(page['ViewportHeight'])
    assert width > 0 and height > 0 and math.isfinite(width + height)
    searches = [item for item in page.get('SearchInputs', []) if item.get('Name') == 'q']
    rectangles = [_rect(item['Bounds']) for item in searches]
    assert any(x < width and y < height and x + w > 0 and y + h > 0 for x, y, w, h in rectangles), 'No actual Bing search input geometry'
    assert page.get('TextRuns'), 'No actual remote text geometry'
    loaded = {}
    for item in page.get('LoadedStyles', []):
        if item.get('Characters', 0) <= 0:
            continue
        # A real redirect receipt keeps both the DOM-requested stylesheet URI
        # and its validated final URI. Unrelated final resources cannot stand
        # in for the page's declared request. Older unredirected receipts may
        # use Address alone, without inventing an implicit redirect.
        requested = item.get('RequestedAddress', item['Address'])
        final = item.get('FinalAddress', item['Address'])
        _url(requested); _url(final)
        assert _same_url(item['Address'], final), 'The stylesheet final address differs from its actual resource receipt'
        loaded[requested] = item
    for declared in page.get('DeclaredStyleLinks', []):
        _url(declared)
        assert declared in loaded, 'A declared external stylesheet did not load: ' + declared
    return {'SearchInputs': searches, 'LoadedStyles': list(loaded.values()),
            'InlineStyleCount': page.get('InlineStyleCount'), 'DeclaredStyleLinks': page.get('DeclaredStyleLinks', []),
            'CssParityVerified': False, 'Meaning': 'Actual resource receipts and element geometry; full Windows CSS/JS parity is not inferred.'}


def native_navigation_failure(serial, after, window_id):
    """Only a fresh complete state for the current native Bing window qualifies."""
    matches = [item for item in BROWSER_STATE.finditer(serial, after) if int(item[1]) == window_id]
    if not matches:
        return None
    item = matches[-1]
    try:
        address = _url(item[2])
    except (AssertionError, TypeError, ValueError):
        return None
    if address.hostname.lower() != 'bing.com' and not address.hostname.lower().endswith('.bing.com'):
        return None
    content = item[3].strip()
    if not content.startswith('Navigation failed:'):
        return None
    return {'WindowId': window_id, 'Address': item[2], 'Text': content,
            'SerialOffset': item.start(), 'Raw': item[0]}


def matching_presentation(serial, page, window_id, after=0):
    """The latest paint for this window must be this exact immutable DOM snapshot."""
    matches = [item for item in PRESENTED.finditer(serial, after) if int(item[1]) == window_id]
    assert matches, 'No actual native browser presentation'
    item = matches[-1]
    assert _same_url(item[3], page['Address']) and int(item[4]) == page['SnapshotSequence'], 'The latest presented browser snapshot differs from the remote DOM'
    assert int(item[5]) > 0 and int(item[6]) > 0, 'No actual scanout presentation identity/time'
    assert abs(float(item[7]) - float(page.get('ScrollY', 0))) <= .001, 'Rendered remote scroll position differs'
    return item


def presentation_identity(item):
    return {'WindowId': int(item[1]), 'Generation': int(item[2]), 'Address': item[3],
            'SnapshotSequence': int(item[4]), 'PresentationSequence': int(item[5]),
            'PresentationTicks': int(item[6]), 'ScrollY': float(item[7])}


def stable_screenshot_presentation(before, after):
    for name in ('WindowId', 'Generation', 'Address', 'SnapshotSequence', 'ScrollY'):
        assert before[name] == after[name], 'Browser presentation changed during screenshot: ' + name
    assert after['PresentationSequence'] >= before['PresentationSequence'] and after['PresentationTicks'] >= before['PresentationTicks'], 'Browser presentation counters regressed during screenshot'
    return {'Before': before, 'After': after, 'StableSnapshot': True,
            'Meaning': 'Actual QMP framebuffer capture bracketed by the same immutable native browser snapshot; this is not physical display timing.'}


def capture_matching_presentation(serial, page, window_id, after, screenshot, healthy, name):
    before = presentation_identity(matching_presentation(serial, page, window_id, after))
    screenshot(name)
    return stable_screenshot_presentation(before,
        presentation_identity(matching_presentation(healthy(), page, window_id, after)))


def _color(value):
    value = value.strip().lower()
    if re.fullmatch(r'#?[0-9a-f]{6}', value):
        value = value.lstrip('#')
        return tuple(int(value[index:index+2], 16) for index in (0, 2, 4))
    match = re.fullmatch(r'rgba?\(\s*(\d+(?:\.\d+)?)\s*[, ]\s*(\d+(?:\.\d+)?)\s*[, ]\s*(\d+(?:\.\d+)?)(?:\s*[,/]\s*[\d.]+)?\s*\)', value)
    assert match, 'Unsupported diagnostic text color: ' + value
    return tuple(round(float(number)) for number in match.groups())


def remote_text_pixels(frame, page, viewport, scale=1):
    """Require nonuniform ink at visible actual text geometry in the guest PNG."""
    image = frame.convert('RGB'); samples = []
    for run in page.get('TextRuns', [])[:128]:
        if len(run.get('Text', '').strip()) < 2:
            continue
        x, y, width, height = _rect(run['Bounds'])
        left, top = max(0, x), max(0, y)
        right, bottom = min(viewport[2], x + width), min(viewport[3], y + height)
        if right <= left or bottom <= top:
            continue
        bounds = [round((viewport[0] + left) * scale), round((viewport[1] + top) * scale),
                  round((viewport[0] + right) * scale), round((viewport[1] + bottom) * scale)]
        region = image.crop(bounds); color = _color(run['Color'])
        ink = [(px, py) for py in range(region.height) for px in range(region.width)
               if all(abs(region.getpixel((px, py))[channel] - color[channel]) <= 14 for channel in range(3))]
        area = region.width * region.height
        columns = len({point[0] for point in ink}); rows = len({point[1] for point in ink})
        passed = len(ink) >= 12 and columns >= 6 and rows >= 3 and len(ink) < area * .75
        samples.append({'Text': run['Text'][:96], 'NativeBounds': bounds, 'InkPixels': len(ink), 'InkColumns': columns, 'InkRows': rows, 'Passed': passed})
    return {'Passed': sum(item['Passed'] for item in samples) >= 2, 'Samples': samples,
            'Meaning': 'Nonuniform text ink inside real DOM rectangles in guest framebuffer PNG; no host browser image or synthetic acceptance pixels.'}


def validate_initialization(page):
    assert page.get('DocumentInitializationComplete') is True and page.get('ScriptBudgetExceeded') is False and page.get('PendingResources') == 0, 'Original page scripts have not finished within their real execution budgets'
    diagnostics = page.get('Diagnostics', '')
    assert not re.search(r'Event loop failed:|Script has allocated|statement count|recursion depth|script execution timed out|operation has timed out|underlying compression routine', diagnostics, re.I), 'Page initialization stopped at a script or decoder error'
    errors = [line for line in diagnostics.splitlines() if (line.startswith(('JavaScript error:', 'Browser event failed:', 'Network callback failed:', 'External script failed:')) or re.match(r'^\w*Error:', line))
              and '_w is not defined' not in line]  # Also observed on the official Chromium reference page.
    assert not errors, 'Page initialization has unresolved JavaScript errors: ' + '; '.join(errors)


def validate_visible_images(page):
    assert isinstance(page.get('MissingVisibleImages'), list), 'Missing actual visible-image readback'
    assert not page['MissingVisibleImages'], 'A visible page image has not decoded: ' + str(page['MissingVisibleImages'])


def validate_citation_layout(page):
    pairs = []
    for identity in {run.get('ListItemId') for run in page.get('TextRuns', []) if run.get('ListItemId')}:
        runs = [run for run in page['TextRuns'] if run.get('ListItemId') == identity]
        cites = [run for run in runs if 'cite' in run.get('Ancestors', '').split('|')]
        # A result title is linked. Unlinked related-topic headings may share
        # the outer result's list item but have independent geometry.
        titles = [run for run in runs if re.search(r'(?:^|\|)a\|(?:(?!h2\|)[^|]+\|)*h2(?:\||$)', run.get('Ancestors', ''))]
        if not cites or not titles: continue
        bottom = max(_rect(run['Bounds'])[1] + _rect(run['Bounds'])[3] for run in cites)
        top = min(_rect(run['Bounds'])[1] for run in titles)
        assert bottom <= top + 1, 'Citation text overlaps its actual result title'
        pairs.append({'ListItemId': identity, 'CitationBottom': bottom, 'TitleTop': top})
    assert pairs, 'No actual citation/title text pair was observed'
    return {'Passed': True, 'Pairs': pairs}


def validate_search_navigation(page):
    links = page.get('NavigationLinks')
    assert isinstance(links, list) and links, 'Missing actual navigation paint readback'
    menus = {}
    for link in links:
        x,y,width,height = _rect(link['ContainerBounds'])
        if 0 <= y < float(page['ViewportHeight']) / 2 and width >= float(page['ViewportWidth']) / 2:
            menus.setdefault(link['MenuId'], []).append(link)
    qualified = [items for items in menus.values() if len(items) >= 5]
    assert qualified, 'No complete actual top search navigation menu'
    for items in qualified:
        assert all(item.get('Painted') is True for item in items), 'A declared search navigation link was clipped or unpainted'
    return {'Passed': True, 'Links': [item for items in qualified for item in items]}


def chevron_pixels(frame, page, viewport):
    arrows = [box for box in page.get('LayoutSamples', []) if 'nav-arrow' in box.get('Class', '').split()]
    assert len(arrows) == 2, 'Actual homepage chevron layout is absent'
    checks = []
    for box in arrows:
        x, y, width, height = _rect(box['Bounds']); matrix = box.get('Transform', [])
        assert 12 < width < 17 and 12 < height < 17 and len(matrix) == 6, 'The arrow kept its unscaled L geometry'
        assert all(abs(abs(matrix[index]) - math.sqrt(.5) / 4) < .002 for index in range(4)), 'The original CSS scale/rotation was not applied'
        region = frame.convert('RGB').crop((math.floor(viewport[0]+x)-1, math.floor(viewport[1]+y)-1,
                                          math.ceil(viewport[0]+x+width)+1, math.ceil(viewport[1]+y+height)+1))
        ink = [(xx, yy) for yy in range(region.height) for xx in range(region.width)
               if min(region.getpixel((xx, yy))) >= 70 and max(region.getpixel((xx, yy))) - min(region.getpixel((xx, yy))) <= 20]
        assert 10 <= len(ink) <= 100, 'The actual framebuffer lacks a thin rotated chevron'
        checks.append({'Bounds': box['Bounds'], 'Transform': matrix, 'InkPixels': len(ink)})
    return {'Passed': True, 'Arrows': checks}


def validate_scroll_coverage(frames):
    assert frames, 'No actual scroll captures'
    first = frames[0]['RemotePage']; covered = 0; previous_sequence = 0
    assert first.get('ScrollY') == 0, 'Full content coverage must start at the top'
    for capture in frames:
        page = capture['RemotePage']; validate_initialization(page); validate_visible_images(page)
        assert page['Address'] == first['Address'] and page['BodySha256'] == first['BodySha256'], 'Scroll coverage crossed a navigation'
        assert page['SnapshotSequence'] >= previous_sequence, 'Scroll snapshots regressed'
        previous_sequence = page['SnapshotSequence']
        top, height = float(page['ScrollY']), float(page['ViewportHeight'])
        assert top <= covered + 1 and height > 0, 'A vertical region was skipped'
        covered = max(covered, top + height)
    assert covered + 2 >= float(frames[-1]['RemotePage']['DocumentHeight']), 'The actual document bottom was not covered'
    return {'Passed': True, 'ReachedBottom': True, 'Frames': frames, 'CoveredThrough': covered,
            'DocumentHeight': frames[-1]['RemotePage']['DocumentHeight']}


def validate_loaded_home(page):
    """A gray response with a couple of controls is explicitly insufficient."""
    assert page.get('PendingResources') == 0, 'The real page still has queued resources'
    width, height = float(page['ViewportWidth']), float(page['ViewportHeight'])
    fields = [field for field in page.get('SearchInputs', []) if field.get('Name') == 'q']
    assert any(150 <= _rect(field['Bounds'])[2] <= width and 12 <= _rect(field['Bounds'])[3] <= 100 for field in fields), 'No usable search box'
    photos = [item for item in page.get('LoadedImages', []) if item.get('Width', 0) >= 600 and item.get('Height', 0) >= 400
              and urlsplit(item.get('Source', '')).scheme == 'https' and len({sample.get('Argb') for sample in item.get('Samples', [])}) >= 8
              and _rect(item['Bounds'])[2] >= width * .7 and _rect(item['Bounds'])[3] >= height * .6]
    assert photos, 'The actual homepage photograph has not decoded and entered the paint list'
    validate_initialization(page)
    return photos


def validate_wide_home(page):
    assert page['ViewportWidth'] >= 1200, 'Bing content acceptance requires the actual maximized viewport'
    navigation = [item for item in page.get('Links', []) if _rect(item['Bounds'])[1] < 100]
    news = [item for item in page.get('Links', []) if urlsplit(item.get('Address', '')).path == '/search'
            and parse_qs(urlsplit(item['Address']).query).get('q') and 180 <= _rect(item['Bounds'])[1] < 450]
    assert len(navigation) >= 4 and len({item['Address'] for item in news}) >= 6, 'Homepage navigation or six actual hot-search links are absent'
    return {'NavigationLinks': navigation, 'HotSearchLinks': news, 'Passed': True}


def photograph_pixels(frame, photos, viewport):
    """Compare actual decoded image samples to actual guest framebuffer pixels."""
    checks = []
    rgb = frame.convert('RGB')
    for photo in photos:
        matched = 0; tested = 0
        for sample in photo.get('Samples', []):
            color = int(sample['Argb'], 16)
            if color >> 24 != 255:
                continue
            x, y = round(viewport[0] + sample['X']), round(viewport[1] + sample['Y'])
            if not (0 <= x < frame.width and 0 <= y < frame.height):
                continue
            expected = ((color >> 16) & 255, (color >> 8) & 255, color & 255)
            actual = rgb.getpixel((x, y)); tested += 1
            matched += max(abs(a - b) for a, b in zip(expected, actual)) <= 6
        checks.append({'Source': photo['Source'], 'Samples': tested, 'Matched': matched,
                       'Passed': tested >= 12 and matched >= max(8, tested * .25)})
    return {'Passed': any(item['Passed'] for item in checks), 'Checks': checks,
            'Meaning': 'Actual decoded source image colors agree with guest framebuffer samples; text-only gray pages fail.'}


def validate_search_results(page, query, link_pattern=None):
    uri = _url(page['Address'])
    assert uri.path == '/search' and parse_qs(uri.query).get('q') == [query], 'Search did not navigate using the entered query'
    assert page.get('PendingResources') == 0, 'Search resources are still queued'
    assert any(item.get('Name') == 'q' and item.get('Value') == query for item in page.get('SearchInputs', [])), 'The actual search input does not retain the query'
    link_pattern = link_pattern or re.compile(r'\.net', re.I)
    links = [item for item in page.get('Links', []) if link_pattern.search(item.get('Text', ''))
             and (urlsplit(item.get('Address', '')).hostname or '').endswith('.microsoft.com')]
    assert links, 'No actual visible Microsoft result link matches the entered query'
    for link in links:
        x, y, width, height = _rect(link['Bounds'])
        assert width > 10 and height > 5 and x < float(page['ViewportWidth']) and y < float(page['ViewportHeight'])
    validate_initialization(page)
    return links


def validate_search_images(page):
    icons = [item for item in page.get('LoadedImages', []) if item.get('Format') == 'PNG'
             and 12 <= item.get('Width', 0) <= 256 and 12 <= item.get('Height', 0) <= 256
             and item.get('Opacity', 1) >= .99 and _rect(item['Bounds'])[2] >= 12
             and len({sample.get('Argb') for sample in item.get('Samples', []) if sample.get('Argb', '')[:2].lower() == 'ff'}) >= 2]
    assert icons, 'Search result images are still placeholders; a decoded small PNG must enter the visible paint list'
    return icons


def run_bing(proof_directory, log, healthy, wait, screenshot, key, type_text, move, *, click=None, wheel=None, timeout=180, open_browser=True, start_offset=None):
    """Open the product's default Browser through QMP. Caller owns QEMU and cleanup."""
    proof_directory = Path(proof_directory)
    started = time.monotonic(); initial_log = healthy()
    assert open_browser or type(start_offset) is int, 'An already-open browser requires the actual input start offset'
    after = len(initial_log) if start_offset is None else start_offset
    assert 0 <= after <= len(initial_log), 'Invalid actual navigation evidence offset'
    result = {'Kind': 'XfeIndependentNativeBingEvidence', 'Passed': False, 'AllUserRequirementsMet': False,
              'NativeHttpsBingLoaded': False, 'FullCssParityVerified': False, 'CompleteJavaScriptCompatibility': False,
              'HostHttpRequests': False, 'HostBrowserScreenshots': False, 'HostJavaScriptInjection': False,
              'VmStartedByThisTool': False, 'Input': 'Actual QMP keyboard/mouse only', 'RequestedUrl': HOME,
              'AcceptanceVersion': 9, 'VisibleContentLoaded': False, 'NavigationVisible': False, 'CssFixesVerified': False, 'VisualHomeLoaded': False, 'RealSearchCompleted': False, 'RepeatedSearchCompleted': False, 'DecodedPngVerified': False, 'SearchImagesLoaded': False,
              'SearchScrolled': False, 'TimeoutSeconds': timeout, 'Screenshots': []}
    output = proof_directory / 'BingEvidence.json'

    def persist():
        result['ElapsedHostSeconds'] = time.monotonic() - started
        output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')

    persist()
    try:
        # The desktop profile starts with its menu closed. Preserve the product
        # default navigation: do not replace it with an offline/failure fixture.
        if open_browser:
            key('ctrl-esc'); key('down'); key('down'); key('ret')
        wait('XFE_DESKTOP_ACTIVE|Browser', after)
        end = time.monotonic() + timeout
        last_error = 'No native response packet'
        accepted = None
        while time.monotonic() < end:
            serial = healthy()
            try:
                window_id, _ = window_geometry(serial, 'Browser')
            except AssertionError:
                window_id = None  # The complete desktop-state packet may still be arriving.
            failure = native_navigation_failure(serial, after, window_id) if window_id is not None else None
            if failure is not None:
                result['NavigationFailure'] = {key: value for key, value in failure.items() if key != 'Raw'}
                raise AssertionError('Actual native Bing navigation failed: ' + failure['Text'])
            try:
                packets = json_blocks(serial, RESPONSE, after)
                chain = validate_response_chain(packets)
                final = chain[-1]['Value']
                remote_packets = json_blocks(serial, REMOTE, after)
                for remote in reversed(remote_packets):
                    page = remote['Value']
                    css = validate_remote_page(page, final)
                    validate_loaded_home(page)
                    validate_wide_home(page)
                    validate_visible_images(page)
                    window_id, bounds = window_geometry(serial, 'Browser')
                    presented = matching_presentation(serial, page, window_id, after)
                    accepted = (chain, remote, css, presented, bounds)
                    break
                assert accepted, 'No matching rendered remote DOM readback'
                break
            except (AssertionError, KeyError, TypeError, ValueError) as error:
                last_error = str(error)
            time.sleep(.1)
        assert accepted, 'Actual native Bing acceptance timed out: ' + last_error
        chain, remote, css, presented, bounds = accepted
        result['Responses'] = [packet['Value'] for packet in chain]
        result['RemotePage'] = remote['Value']; result['CssResources'] = css
        result['Presentation'] = presentation_identity(presented)
        result['HomeContent'] = validate_wide_home(remote['Value'])
        viewport = [bounds[0] + 2, bounds[1] + 67, bounds[2] - 4, bounds[3] - 96]
        assert abs(viewport[2] - float(remote['Value']['ViewportWidth'])) <= 1 and abs(viewport[3] - float(remote['Value']['ViewportHeight'])) <= 1
        move(1200, 710)
        for attempt in range(1, 5):
            serial = healthy(); window_id, current_bounds = window_geometry(serial, 'Browser')
            assert current_bounds == bounds, 'The browser geometry changed before its actual screenshot'
            before = presentation_identity(matching_presentation(serial, remote['Value'], window_id, after))
            name = 'Browser-Bing-Native-' + str(attempt)
            screenshot(name)
            serial = healthy(); current_id, current_bounds = window_geometry(serial, 'Browser')
            assert current_id == window_id and current_bounds == bounds, 'The browser geometry changed during its actual screenshot'
            presentation = stable_screenshot_presentation(before,
                presentation_identity(matching_presentation(serial, remote['Value'], window_id, after)))
            path = proof_directory / (name + '.png')
            with Image.open(path) as frame:
                assert frame.width >= viewport[0] + viewport[2] and frame.height >= viewport[1] + viewport[3]
                ink = remote_text_pixels(frame, remote['Value'], viewport)
                photo = photograph_pixels(frame, validate_loaded_home(remote['Value']), viewport)
                home_chevrons = chevron_pixels(frame, remote['Value'], viewport)
                hot_search_rects = [_rect(item['Bounds']) for item in result['HomeContent']['HotSearchLinks']]
                runs = [run for run in remote['Value']['TextRuns'] if any(abs(_rect(run['Bounds'])[1] - rect[1]) <= rect[3] for rect in hot_search_rects)]
                content_pixels = remote_text_pixels(frame, {**remote['Value'], 'TextRuns': runs}, viewport)
            result['Screenshots'].append({'Path': path.name, 'Sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
                'BodySha256': remote['Value']['BodySha256'], 'PresentationCorrelation': presentation, 'TextPixels': ink, 'PhotographPixels': photo, 'HomeContentPixels': content_pixels})
            if ink['Passed'] and photo['Passed'] and content_pixels['Passed']:
                break
            time.sleep(.25)
        assert result['Screenshots'][-1]['TextPixels']['Passed'], 'The actual Bing text was not painted at its remote DOM geometry'
        assert result['Screenshots'][-1]['PhotographPixels']['Passed'], 'The actual homepage photograph was not painted'
        assert result['Screenshots'][-1]['HomeContentPixels']['Passed'], 'The actual homepage hot-search content was not painted'
        result['HomeChevronPixels'] = home_chevrons
        result.update(NativeHttpsBingLoaded=True, VisualHomeLoaded=True)
        assert click is not None, 'Real search requires actual QMP pointer input'
        query = 'Microsoft .NET'
        field = next(item for item in remote['Value']['SearchInputs'] if item['Name'] == 'q' and item['Bounds']['Width'] >= 150)
        rect = _rect(field['Bounds']); search_after = len(healthy())
        click(round(viewport[0] + rect[0] + 24), round(viewport[1] + rect[1] + rect[3] / 2))
        key('ctrl-a'); type_text(query); key('ret')
        search_accepted = None; search_end = time.monotonic() + timeout; first_search_candidate = None; search_capture_attempts = 0
        while time.monotonic() < search_end:
            serial = healthy()
            try:
                packets = json_blocks(serial, RESPONSE, search_after)
                initial = next(packet['Value']['RequestUri'] for packet in packets
                               if urlsplit(packet['Value'].get('RequestUri', '')).path == '/search'
                               and parse_qs(urlsplit(packet['Value']['RequestUri']).query).get('q') == [query])
                pattern = re.compile(re.escape(query), re.I)
                search_chain = validate_response_chain(packets, initial, pattern); final = search_chain[-1]['Value']
                for packet in reversed(json_blocks(serial, REMOTE, search_after)):
                    page = packet['Value']; search_css = validate_remote_page(page, final, pattern)
                    links = validate_search_results(page, query)
                    validate_search_images(page)
                    validate_visible_images(page)
                    validate_citation_layout(page);validate_search_navigation(page)
                    if first_search_candidate is None: first_search_candidate = time.monotonic()
                    assert time.monotonic() - first_search_candidate >= 3, 'Waiting for initial search callbacks to finish'
                    current_id, current_bounds = window_geometry(serial, 'Browser')
                    assert current_bounds == bounds
                    current_presentation = matching_presentation(serial, page, current_id, search_after)
                    if search_capture_attempts == 8: raise RuntimeError('No stable first-search screenshot after eight bounded attempts')
                    search_capture_attempts += 1
                    capture_name = 'Browser-Bing-Search-Native' + ('' if search_capture_attempts == 1 else '-Attempt-' + str(search_capture_attempts))
                    path = proof_directory / (capture_name + '.png')
                    result.setdefault('SearchCaptureAttempts', []).append({'Path': path.name, 'SnapshotSequence': page['SnapshotSequence'], 'Accepted': False})
                    presentation = capture_matching_presentation(serial, page, current_id, search_after, screenshot, healthy, capture_name)
                    result['SearchCaptureAttempts'][-1]['Accepted'] = True
                    search_accepted = (search_chain, page, search_css, links, current_presentation, presentation, path); break
                assert search_accepted, 'No complete search snapshot'
                break
            except (AssertionError, KeyError, TypeError, ValueError, StopIteration) as error:
                last_error = str(error)
            time.sleep(.1)
        assert search_accepted, 'Actual search timed out: ' + last_error
        search_chain, page, search_css, links, current_presentation, presentation, path = search_accepted
        with Image.open(path) as frame:
            ink = remote_text_pixels(frame, page, viewport)
            icon_pixels = photograph_pixels(frame, validate_search_images(page), viewport)
        assert ink['Passed'], 'The actual search results did not produce visible text'
        assert icon_pixels['Passed'], 'Decoded search PNG images were not actually painted'
        result['Search'] = {'Query': query, 'Responses': [item['Value'] for item in search_chain], 'RemotePage': page,
                            'VisibleResultLinks': links, 'CssResources': search_css, 'CitationLayout': validate_citation_layout(page), 'Navigation':validate_search_navigation(page),
                            'Screenshot': {'Path': path.name, 'Sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
                                'BodySha256': page['BodySha256'], 'PresentationCorrelation': presentation, 'TextPixels': ink, 'IconPixels': icon_pixels}}
        result['RealSearchCompleted'] = True
        result['SearchImagesLoaded'] = True
        assert 'PNG' in set(result['RemotePage'].get('DecodedImageFormats', [])) | set(page.get('DecodedImageFormats', [])), 'No actual Bing PNG asset decoded in the native runtime'
        result['DecodedPngVerified'] = True
        wheel_unit=40
        def scan_content(record, current, initial_frames, prefix):
            captures=list(initial_frames)
            for index in range(1,33):
                if float(current['ScrollY'])+float(current['ViewportHeight'])+2>=float(current['DocumentHeight']):break
                scroll_after=len(healthy());previous_y=float(current['ScrollY']);wheel(max(1,min(10,int(float(current['ViewportHeight'])*.6/wheel_unit))))
                deadline=time.monotonic()+timeout;next_page=None
                while time.monotonic()<deadline:
                    serial=healthy()
                    for packet in reversed(json_blocks(serial,REMOTE,scroll_after)):
                        item=packet['Value']
                        if item.get('Address')!=current['Address'] or float(item.get('ScrollY',0))<=previous_y:continue
                        try:
                            validate_initialization(item);validate_visible_images(item)
                            shown=matching_presentation(serial,item,current_id,scroll_after)
                            next_page=item;break
                        except AssertionError:pass
                    if next_page:break
                    time.sleep(.1)
                assert next_page,'Actual content scrolling did not present a later complete frame'
                current=next_page;before=presentation_identity(shown);name=prefix+'-'+str(index);screenshot(name)
                stable=stable_screenshot_presentation(before,presentation_identity(matching_presentation(healthy(),current,current_id,scroll_after)))
                path=proof_directory/(name+'.png')
                with Image.open(path) as frame:ink=remote_text_pixels(frame,current,viewport)
                assert ink['Passed'],'An actually scrolled content region is blank'
                captures.append({'RemotePage':current,'Screenshot':{'Path':path.name,'Sha256':hashlib.sha256(path.read_bytes()).hexdigest(),
                    'BodySha256':current['BodySha256'],'PresentationCorrelation':stable,'TextPixels':ink}})
            record['FullContent']=validate_scroll_coverage(captures)
            return current

        if wheel is not None:
            scroll_after = len(healthy()); wheel(6)
            scroll_end = time.monotonic() + timeout; scrolled = None
            while time.monotonic() < scroll_end:
                serial = healthy()
                for packet in reversed(json_blocks(serial, REMOTE, scroll_after)):
                    candidate = packet['Value']
                    if candidate.get('Address') != page['Address'] or candidate.get('ScrollY', 0) < page.get('ScrollY', 0) + 50:
                        continue
                    try:
                        validate_search_results(candidate, query)
                        shown = matching_presentation(serial, candidate, current_id, scroll_after)
                        scrolled = (candidate, shown); break
                    except AssertionError: pass
                if scrolled: break
                time.sleep(.1)
            assert scrolled, 'Actual search wheel input did not scroll and present results'
            candidate, shown = scrolled; before = presentation_identity(shown)
            screenshot('Browser-Bing-Search-Scrolled-Native')
            presentation = stable_screenshot_presentation(before, presentation_identity(matching_presentation(healthy(), candidate, current_id, scroll_after)))
            path = proof_directory / 'Browser-Bing-Search-Scrolled-Native.png'
            with Image.open(path) as frame: ink = remote_text_pixels(frame, candidate, viewport)
            assert ink['Passed'], 'Scrolled results were blank'
            result['Search']['ScrolledScreenshot'] = {'Path': path.name, 'Sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
                'ScrollY': candidate['ScrollY'], 'BodySha256': candidate['BodySha256'], 'PresentationCorrelation': presentation, 'TextPixels': ink}
            result['Search']['ScrolledRemotePage'] = candidate
            result['SearchScrolled'] = True
        wheel_unit=(float(candidate['ScrollY'])-float(page.get('ScrollY',0)))/6
        assert wheel_unit>0,'No measured wheel displacement'
        current=scan_content(result['Search'],candidate,[{'RemotePage':page,'Screenshot':result['Search']['Screenshot']},
            {'RemotePage':candidate,'Screenshot':result['Search']['ScrolledScreenshot']}],'Browser-Bing-Search-Full')
        # The first query alone previously missed a NodeList/navigation failure
        # in the initialized result page. Exercise a second real user query.
        assert wheel is not None, 'Repeated search requires actual wheel input'
        restore_after=len(healthy());wheel(-max(6,math.ceil(float(current['ScrollY'])/wheel_unit)+6));restored=None;restore_end=time.monotonic()+timeout
        while time.monotonic()<restore_end:
            serial=healthy()
            for packet in reversed(json_blocks(serial,REMOTE,restore_after)):
                candidate=packet['Value']
                if candidate.get('Address')==page['Address'] and candidate.get('ScrollY',1)==0:
                    try:
                        validate_search_results(candidate,query);matching_presentation(serial,candidate,current_id,restore_after)
                        restored=candidate;break
                    except AssertionError:pass
            if restored:break
            time.sleep(.1)
        assert restored,'Actual wheel input did not restore the search box'
        field=next(item for item in restored['SearchInputs'] if item['Name']=='q');rect=_rect(field['Bounds'])
        repeated_query='Microsoft CSharp';repeated_after=len(healthy())
        click(round(viewport[0]+rect[0]+24),round(viewport[1]+rect[1]+rect[3]/2));key('ctrl-a');type_text(repeated_query);key('ret')
        repeated=None;repeated_end=time.monotonic()+timeout;first_candidate=None;repeated_capture_attempts=0
        pattern=re.compile(r'csharp|c#',re.I)
        while time.monotonic()<repeated_end:
            serial=healthy()
            try:
                packets=json_blocks(serial,RESPONSE,repeated_after)
                initial=next(packet['Value']['RequestUri'] for packet in packets if urlsplit(packet['Value'].get('RequestUri','')).path=='/search' and parse_qs(urlsplit(packet['Value']['RequestUri']).query).get('q')==[repeated_query])
                chain=validate_response_chain(packets,initial,pattern);final=chain[-1]['Value']
                for packet in reversed(json_blocks(serial,REMOTE,repeated_after)):
                    candidate=packet['Value'];css=validate_remote_page(candidate,final,pattern)
                    repeated_links=validate_search_results(candidate,repeated_query,pattern)
                    validate_visible_images(candidate);validate_citation_layout(candidate);validate_search_navigation(candidate);validate_search_images(candidate)
                    if first_candidate is None:first_candidate=time.monotonic()
                    assert time.monotonic()-first_candidate>=3,'Waiting for repeated-search initialization'
                    shown=matching_presentation(serial,candidate,current_id,repeated_after)
                    if repeated_capture_attempts == 8: raise RuntimeError('No stable repeated-search screenshot after eight bounded attempts')
                    repeated_capture_attempts += 1
                    capture_name = 'Browser-Bing-SecondSearch-Native' + ('' if repeated_capture_attempts == 1 else '-Attempt-' + str(repeated_capture_attempts))
                    path = proof_directory / (capture_name + '.png')
                    result.setdefault('RepeatedSearchCaptureAttempts', []).append({'Path': path.name, 'SnapshotSequence': candidate['SnapshotSequence'], 'Accepted': False})
                    presentation=capture_matching_presentation(serial,candidate,current_id,repeated_after,screenshot,healthy,capture_name)
                    result['RepeatedSearchCaptureAttempts'][-1]['Accepted']=True
                    repeated=(chain,candidate,css,repeated_links,shown,presentation,path);break
                if repeated:break
            except (AssertionError,KeyError,TypeError,ValueError,StopIteration) as error:last_error=str(error)
            time.sleep(.1)
        assert repeated,'Repeated actual search timed out: '+last_error
        chain,candidate,css,repeated_links,shown,presentation,path=repeated
        with Image.open(path) as frame:
            ink=remote_text_pixels(frame,candidate,viewport);repeated_icons=photograph_pixels(frame,validate_search_images(candidate),viewport)
        assert repeated_icons['Passed'],'Repeated-search icons have not entered the actual native frame'
        assert ink['Passed'],'The repeated search did not paint actual result text'
        result['RepeatedSearch']={'Query':repeated_query,'Responses':[packet['Value'] for packet in chain],'RemotePage':candidate,'CssResources':css,'VisibleResultLinks':repeated_links,'CitationLayout':validate_citation_layout(candidate),'Navigation':validate_search_navigation(candidate),
            'Screenshot':{'Path':path.name,'Sha256':hashlib.sha256(path.read_bytes()).hexdigest(),'BodySha256':candidate['BodySha256'],'PresentationCorrelation':presentation,'TextPixels':ink,'IconPixels':repeated_icons}}
        scan_content(result['RepeatedSearch'],candidate,[{'RemotePage':candidate,'Screenshot':result['RepeatedSearch']['Screenshot']}],'Browser-Bing-SecondSearch-Full')
        result['NavigationVisible']=True
        result['VisibleContentLoaded']=True
        result['CssFixesVerified']=True
        result['RepeatedSearchCompleted']=True
        result.update(Passed=True,
                      Meaning='Actual guest HTTPS homepage, rotated chevrons, nonoverlapping citations and titles, and overlapping framebuffer captures through the bottom of both typed search pages; complete browser compatibility is not inferred.')
    except Exception as error:
        result['Failure'] = str(error)
        raise
    finally:
        # Raw packets survive timeout and partial failure, without dumping the
        # entire HTML body or making another network request on the host.
        try:
            serial = log()
            packets = json_blocks(serial, RESPONSE, after) + json_blocks(serial, REMOTE, after)
            raw = proof_directory / 'BingRequestEvidence.log'
            presented_packets = [{'Raw': match[0], 'SerialOffset': match.start()} for match in PRESENTED.finditer(serial, after)]
            failure_packets = [{'Raw': match[0], 'SerialOffset': match.start()} for match in BROWSER_STATE.finditer(serial, after)
                               if match[3].strip().startswith('Navigation failed:')]
            raw.write_text('\n'.join(packet['Raw'] for packet in sorted(packets + presented_packets + failure_packets, key=lambda item: item['SerialOffset'])), encoding='utf-8')
            result['RawGuestEvidence'] = {'Path': raw.name, 'Sha256': hashlib.sha256(raw.read_bytes()).hexdigest(), 'Packets': len(packets)}
            result['RawGuestEvidence']['PresentationMarkers'] = len(presented_packets)
            result['RawGuestEvidence']['NavigationFailures'] = len(failure_packets)
        except Exception as error:
            result['RawGuestEvidenceFailure'] = str(error)
            result['Passed'] = False
        persist()
    return result