XFESystem

"""Independent-kernel document acceptance using physical PS/2 input and scanout.

The shared helpers below only classify pixels or validate read-only DOM data.
The former Cosmos boot harness and its receipts are never executed or reused.
"""
import hashlib
import json
from pathlib import Path
import re
import textwrap
import time

from PIL import Image, ImageChops
from NativeCSharpEvidence import normalized_color
from TestDesktopDocuments import flat_article_pixels, reading_progress_pixels
from TestWebsiteDocument import DOCUMENT, paint_evidence, verify_base

ROOT = Path(__file__).resolve().parents[2]
DOCUMENT_PATH = r'X:\Users\Default\Documents\Readme.md'
LONG_PATH = r'X:\Users\Default\Documents\WebsiteParity.md'
MODEL = re.compile(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END', re.S)
STATE = re.compile(r'XFE_BROWSER_STATE_BEGIN\r?\n(\d+)\|([^\r\n]+)\r?\n(.*?)\r?\nXFE_BROWSER_STATE_END', re.S)
WINDOW = re.compile(r'^XFE_WINDOW_STATE\|(\d+)\|(\w+)\|(-?\d+)\|(-?\d+)\|(\d+)\|(\d+)\|([01])\|([01])\|([01])$', re.M)
MODE = re.compile(r'^XFE_WINDOW_DOCUMENT\|(\d+)\|([^|\r\n]+)\|(Preview|Edit|Split)\|([01])\|([01])\|([01])\|([^\r\n]*)$', re.M)
PRESENTED = re.compile(r'^XFE_BROWSER_BACKGROUND_PRESENTED\|Window\|(\d+)\|Generation\|(\d+)\|Address\|([^|\r\n]+)'
                       r'\|SnapshotSequence\|(\d+)\|PresentationSequence\|(\d+)\|PresentationTicks\|(\d+)\|ScrollY\|([\d.]+)$', re.M)


def seed_documents(created_text_file=False):
    """Derive clicked rows from source-bound RAM seeds, then require guest paths."""
    names, sources = set(), {}
    program = None
    for relative in ('Source/XFESystem.Core/MemoryWorkspace.cs', 'Source/XFESystem.NativeRuntime/Application/Program.cs'):
        payload = (ROOT / relative).read_bytes()
        source = payload.decode('utf-8-sig')
        sources[relative] = hashlib.sha256(payload).hexdigest()
        names.update(re.findall(r'Write(?:Text|Bytes)\(SystemPaths\.Combine\(SystemPaths\.Documents,\s*"([^"\r\n]+)"\)', source))
        if relative.endswith('Program.cs'):
            program = source
    if created_text_file:
        names.add('Document1.txt')
    names = sorted(names, key=str.casefold)
    assert {'Readme.md', 'WebsiteParity.md', 'ManagedHello.dll'}.issubset(names), names
    match = re.search(r'WriteText\(SystemPaths\.Combine\(SystemPaths\.Documents,\s*"Readme.md"\),\s*"""\r?\n(.*?)\r?\n[ \t]*"""', program, re.S)
    assert match, 'The source-bound Markdown RAM seed is unavailable'
    initial = textwrap.dedent(match.group(1)).replace('\r', '')
    return {'Names': names, 'SourceSha256': sources, 'InitialReadmeSource': initial,
            'Meaning': 'Rows are derived from the current RAM seed; clicked paths are independently confirmed by guest state.'}


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 run_documents(proof_directory, log, healthy, wait, screenshot, key, type_text, move, click, wheel):
    result = {'Kind': 'XfeIndependentDesktopDocumentWorkflow', 'Passed': False, 'NoRenderFailures': False,
              'Input': 'Actual QMP PS2 keyboard/mouse packets only', 'Screenshots': {},
              'Seed': seed_documents(True), 'HostJavaScriptInjection': False, 'HostScreenshotSubstitution': False}
    output = proof_directory / 'DocumentWorkflow.json'
    started = time.monotonic()

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

    def wait_mode(after, window_id, mode, dirty=None, seconds=180):
        deadline = time.monotonic() + seconds
        last = None
        while time.monotonic() < deadline:
            for match in reversed(list(MODE.finditer(healthy(), after))):
                if int(match[1]) == window_id:
                    last = match.groups()
                    if match[3] == mode and (dirty is None or int(match[4]) == dirty) and match[5] == match[6] == '0':
                        return {'WindowId': window_id, 'Path': match[2], 'Mode': mode, 'Dirty': int(match[4]),
                                'EditorPreparing': False, 'PreviewPreparing': False, 'Status': match[7], 'SerialOffset': match.start()}
            time.sleep(.05)
        raise AssertionError(('Actual document mode did not complete', mode, last, healthy()[-4096:]))

    def open_file(name):
        files_id, bounds = window_geometry(healthy(), 'Files')
        before = len(healthy())
        click(bounds[0] + 70, bounds[1] + 110 + 28 * result['Seed']['Names'].index(name))
        wait('XFE_DESKTOP_ACTIVE|Editor', before)
        wait('XFE_DESKTOP_STATE_END', before)
        window_id, editor_bounds = window_geometry(healthy(), 'Editor')
        path = DOCUMENT_PATH if name == 'Readme.md' else LONG_PATH
        mode = wait_mode(before, window_id, 'Preview', 0)
        assert mode['Path'] == path, ('The physical file row opened another guest path', mode)
        return before, window_id, editor_bounds, files_id

    def close_editor():
        before = len(healthy()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', before)

    def viewport(bounds, split=False):
        x, y, width, height = bounds
        left = (width - 8) // 2 if split else 0
        return [x + 2 + (left + 4 if split else 0), y + 67,
                width - 4 - (left + 4 if split else 0), height - 96]

    def document(after, title, predicate=lambda item: True, seconds=180):
        deadline = time.monotonic() + seconds
        last_error = 'No read-only document proof'
        while time.monotonic() < deadline:
            for match in reversed(list(DOCUMENT.finditer(healthy(), after))):
                try:
                    item = json.loads(match[1])
                    if item.get('Title', '').endswith(title) and predicate(item):
                        verify_base(item)
                        return item, match.end()
                except (ValueError, AssertionError, KeyError, TypeError) as error:
                    last_error = str(error)
            time.sleep(.1)
        raise AssertionError('Actual document readback did not stabilize: ' + last_error + '\n' + healthy()[-4096:])

    def presented(after, window_id, item, previous_sequence=0, expected_scroll=None):
        deadline = time.monotonic() + 90
        while time.monotonic() < deadline:
            for match in PRESENTED.finditer(healthy(), after):
                if int(match[1]) == window_id and match[3] == item['Address'] and int(match[5]) > previous_sequence and int(match[6]) > 0 and \
                    (expected_scroll is None or abs(float(match[7]) - expected_scroll) < 1):
                    # The readback is emitted on the page owner before publishing
                    # its immutable snapshot. An actual UI Present after its END
                    # is necessary, then scanout pixels must independently match.
                    return {'Passed': True, 'WindowId': window_id, 'Address': match[3], 'SnapshotSequence': int(match[4]),
                            'PresentationSequence': int(match[5]), 'PresentationTicks': int(match[6]),
                            'ScrollY': float(match[7]), 'DocumentEndOffset': after, 'SerialOffset': match.start()}
            time.sleep(.1)
        raise AssertionError('No actual scanout presentation followed the matching document readback')

    def capture(name, after, window_id, bounds, title, split=False, contains=None, previous=None, heading=True, different_from=None, ready=None):
        pane = viewport(bounds, split)
        def matching(item):
            return (contains is None or contains in item.get('BodyText', '')) and abs(item['Article']['Bounds']['Width'] - pane[2]) < 1 and \
                (previous is None or item['Article']['Bounds']['Y'] < previous['Document']['Article']['Bounds']['Y'] - 30) and \
                (ready is None or ready(item))
        item, end = document(after, title, matching)
        expected_scroll = previous['Presentation']['ScrollY'] + previous['Document']['Article']['Bounds']['Y'] - item['Article']['Bounds']['Y'] if previous else None
        ack = presented(end, window_id, item, previous['Presentation']['PresentationSequence'] if previous else 0, expected_scroll)
        move(1200, 710)
        deadline = time.monotonic() + 60
        attempt = 0
        while time.monotonic() < deadline:
            healthy(); attempt += 1
            screenshot(name)
            with Image.open(proof_directory / (name + '.png')) as frame:
                pixels = paint_evidence(frame, item, pane, 1, require_heading=True) if heading else None
                flat = flat_article_pixels(frame, item, pane, 1)
                progress = reading_progress_pixels(frame, pane, 1)
                changed = None
                if previous is not None:
                    x, y, width, height = pane
                    with Image.open(proof_directory / previous['Screenshot']) as before:
                        changed = ImageChops.difference(frame.convert('RGB').crop((x, y, x+width, y+height)),
                            before.convert('RGB').crop((x, y, x+width, y+height))).getbbox()
                phrase_pixels = None
                if contains is not None:
                    paragraph = next((part for part in item.get('Paragraphs', []) if contains in part['Text']), None)
                    assert paragraph is not None, 'The new preview phrase has no actual paragraph geometry'
                    rect = paragraph['Bounds']
                    rgb = tuple(int(normalized_color(paragraph['Style']['Color'])[index:index+2], 16) for index in (0, 2, 4))
                    px = [pane[0] + max(0, int(rect['X'])), pane[1] + max(0, int(rect['Y'])),
                          pane[0] + min(pane[2], int(rect['X'] + rect['Width']) + 1), pane[1] + min(pane[3], int(rect['Y'] + rect['Height']) + 1)]
                    region = frame.convert('RGB').crop(px)
                    ink = [(x, y) for y in range(region.height) for x in range(region.width)
                           if all(abs(region.getpixel((x,y))[channel] - rgb[channel]) <= 12 for channel in range(3))]
                    new_ink = ink
                    if different_from:
                        with Image.open(proof_directory / different_from) as before:
                            old = before.convert('RGB').crop(px)
                            new_ink = [(x,y) for x,y in ink if not all(abs(old.getpixel((x,y))[channel] - rgb[channel]) <= 12 for channel in range(3))]
                    phrase_pixels = {'Passed': len(ink) >= 35 and len({x for x,y in ink}) >= 16 and
                                      (different_from is None or len(new_ink) >= 20 and len({x for x,y in new_ink}) >= 12),
                                     'Text': contains, 'MatchingGlyphPixels': len(ink), 'InkColumns': len({x for x,y in ink}),
                                     'NewInkPixels': len(new_ink), 'NewInkColumns': len({x for x,y in new_ink}), 'NativeBounds': px,
                                     'ComparedWith': different_from}
            grew = previous is None or progress.get('NativeFillWidth', 0) >= previous['ReadingProgressPixels']['NativeFillWidth'] + 8
            if (not heading or pixels['Passed']) and flat['Passed'] and progress['Passed'] and grew and (previous is None or changed) and (phrase_pixels is None or phrase_pixels['Passed']):
                path = name + '.png'
                result['Screenshots'][path] = hashlib.sha256((proof_directory / path).read_bytes()).hexdigest()
                return {'Passed': True, 'Document': item, 'Presentation': ack, 'Viewport': pane,
                        'PixelEvidence': pixels, 'FlatArticlePixels': flat, 'ReadingProgressPixels': progress,
                        'PhraseGlyphPixels': phrase_pixels,
                        'ViewportDifferenceBounds': changed, 'Screenshot': path, 'Attempts': attempt}
            time.sleep(.25)
        raise AssertionError(('Actual document pixels did not match', name, pixels, flat, progress, grew, phrase_pixels))

    def wait_model(after, expected=None):
        deadline = time.monotonic() + 180
        latest = None
        while time.monotonic() < deadline:
            for match in reversed(list(MODEL.finditer(healthy(), after))):
                latest = match[1].replace('\r', '')
                if expected is None or latest == expected:
                    return latest, {'ExactModelReadback': True, 'Characters': len(latest),
                        'Utf8Sha256': hashlib.sha256(latest.encode('utf-8')).hexdigest(), 'SerialOffset': match.start()}
            time.sleep(.1)
        raise AssertionError(('The actual worker-owned editor model differs', expected, latest))

    def background_owner(after, window_id, kind):
        pattern = re.compile(r'^XFE_EDITOR_BACKGROUND\|WorkBegin\|Window\|' + str(window_id) +
            r'\|Generation\|(\d+)\|UiThread\|(\d+)\|Thread\|(\d+)\|Kind\|' + kind + r'\|', re.M)
        matches = list(pattern.finditer(healthy(), after))
        assert matches and all(part[2] != part[3] for part in matches), 'The actual page did not have an independent background owner'
        return [{'Generation': int(part[1]), 'UiThread': int(part[2]), 'WorkerThread': int(part[3]), 'SerialOffset': part.start()} for part in matches]

    def save(after, expected):
        key('ctrl-s')
        wait('XFE_DESKTOP_SAVED|' + DOCUMENT_PATH + '|' + str(len(expected)) + '\n', after, seconds=180)
        readback = 'XFE_DESKTOP_SAVED_TEXT_BEGIN\n' + expected[:512] + '\nXFE_DESKTOP_SAVED_TEXT_END'
        wait(readback, after, seconds=180)
        return {'Passed': True, 'Path': DOCUMENT_PATH, 'Characters': len(expected), 'ExactRamReadback': True,
                'Utf8Sha256': hashlib.sha256(expected.encode('utf-8')).hexdigest(), 'Readback': expected[:512]}

    try:
        assert re.search(r'XFE_INPUT_READY keyboard=1 mouse=1 wheel=1 id=3', healthy()), 'Real IntelliMouse ID3 wheel initialization is required'
        assert re.search(r'XFE_INDEPENDENT_INPUT\|[^\r\n]*\|Wheel\|1(?:\||$)', healthy()), 'Managed input did not preserve negotiated wheel capability'
        before, window_id, bounds, files_id = open_file('Readme.md')
        result['DefaultPreview'] = capture('Documents-DefaultPreview', before, window_id, bounds, 'Readme.md')
        assert 'XFE_BROWSER_SCRIPT_BEGIN|0.52.2/Core/EditorCore.js|' not in healthy()[before:], 'Default preview initialized an editor core'
        result['DefaultPreview'].update(WindowId=window_id, GuestPath=DOCUMENT_PATH, MonacoInitializedBeforeEdit=False,
                                       BackgroundOwners=background_owner(before, window_id, 'Document'))
        result['FlatPreview'] = {'Passed': True, **result['DefaultPreview']['FlatArticlePixels']}
        persist()

        before = len(healthy()); click(bounds[0] + bounds[2] - 50, bounds[1] + 49)
        wait('XFE_EDITOR_BACKGROUND|Ready|Window|' + str(window_id) + '|', before, seconds=180)
        initial, model = wait_model(before, result['Seed']['InitialReadmeSource'])
        mode = wait_mode(before, window_id, 'Edit', 0)
        # Plaintext Monaco still needs real source glyphs, rather than a ready
        # model paired with a blank or stale preview framebuffer.
        move(1200, 710)
        deadline = time.monotonic() + 60; source_pixels = None
        while time.monotonic() < deadline:
            screenshot('Documents-Edit')
            pane = viewport(bounds); x, y, width, height = pane
            with Image.open(proof_directory / 'Documents-Edit.png') as frame:
                region = frame.convert('RGB').crop((x+50,y,x+width-20,y+min(height,240)))
                ink = [(ix,iy) for iy in range(region.height) for ix in range(region.width)
                       if min(region.getpixel((ix,iy))) >= 150 and max(region.getpixel((ix,iy))) < 245 and
                       max(region.getpixel((ix,iy))) - min(region.getpixel((ix,iy))) <= 25]
            source_pixels = {'Passed': len(ink) >= 100 and len({ix for ix,iy in ink}) >= 30 and len({iy for ix,iy in ink}) >= 10,
                             'GlyphPixels': len(ink), 'InkColumns': len({ix for ix,iy in ink}), 'InkRows': len({iy for ix,iy in ink})}
            if source_pixels['Passed']: break
            time.sleep(.25)
        assert source_pixels['Passed'], 'The actual plaintext editor has no visible source glyph distribution'
        result['Screenshots']['Documents-Edit.png'] = hashlib.sha256((proof_directory / 'Documents-Edit.png').read_bytes()).hexdigest()
        result['Edit'] = {'Passed': True, 'Mode': mode, 'Model': model, 'ActualReadback': initial, 'SameWindow': True,
                          'SourceGlyphPixels': source_pixels, 'BackgroundOwners': background_owner(before, window_id, 'Editor')}
        persist()

        # Use the existing physical maximize shortcut so the preserved Chinese
        # seed and its new paragraph remain visibly readable in both panes.
        before = len(healthy()); key('meta_l-up')
        deadline = time.monotonic() + 30
        while time.monotonic() < deadline:
            matches = [part for part in WINDOW.finditer(healthy(), before) if int(part[1]) == window_id and part[8] == '1']
            if matches: break
            time.sleep(.05)
        assert matches, 'The actual maximize shortcut did not update document geometry'
        active_id, bounds = window_geometry(healthy(), 'Editor')
        assert active_id == window_id, 'Maximize changed the document window identity'
        before = len(healthy()); click(bounds[0] + bounds[2] - 148, bounds[1] + 49)
        wait_mode(before, window_id, 'Split', 0)
        split = capture('Documents-Split', before, window_id, bounds, 'Readme.md', split=True)
        click(bounds[0] + 100, bounds[1] + 170)
        before = len(healthy()); key('ctrl-end'); type_text('\n\nnative split preview')
        expected = initial + '\n\nnative split preview'
        _, edited = wait_model(before, expected)
        mode = wait_mode(before, window_id, 'Split', 1)
        changed = capture('Documents-Split-Edited', before, window_id, bounds, 'Readme.md', split=True, contains='native split preview', different_from='Documents-Split.png')
        assert changed['Document'] != split['Document'], 'Unsaved preview did not follow the actual model'
        result['Split'] = {'Passed': True, 'SameWindow': True, 'Mode': mode, 'Initial': split, 'Unsaved': changed, 'EditedModel': edited,
                           'UnsavedPreviewFollowsInput': True, 'PhysicalMaximizeShortcut': True}
        result['Save'] = save(len(healthy()), expected)
        persist(); close_editor()

        before, reopened_id, reopened_bounds, _ = open_file('Readme.md')
        result['Reopen'] = capture('Documents-Reopened', before, reopened_id, reopened_bounds, 'Readme.md', contains='native split preview')
        result['Reopen'].update(NewWindow=reopened_id != window_id, DefaultPreview=True, SavedRamTextObserved=True)
        # Reopen the actual source too. A preview/body marker alone cannot prove
        # complete saved text or the absence of a stale editor model.
        before = len(healthy()); click(reopened_bounds[0] + reopened_bounds[2] - 50, reopened_bounds[1] + 49)
        wait('XFE_EDITOR_BACKGROUND|Ready|Window|' + str(reopened_id) + '|', before, seconds=180)
        _, reopened_model = wait_model(before, expected)
        result['Reopen']['ExactReopenedEditorModel'] = reopened_model
        result['Reopen']['SecondSaveRamReadback'] = save(len(healthy()), expected)
        close_editor(); persist()

        before, long_id, long_bounds, _ = open_file('WebsiteParity.md')
        def fallback_ready(item):
            activity = healthy()[before:]
            return '图表组件加载失败' in (item.get('MermaidState') or {}).get('Text', '') and \
                any(line.startswith('XFE_BROWSER_SCRIPT_REJECTED|') and 'mermaid.min.js' in line and
                    '|Reason|ResourceBudget' in line for line in activity.splitlines())
        # The real initial snapshot can precede the asynchronous rejection.
        # Wait for the actual page's fallback plus its native admission record,
        # then require that matching snapshot's presentation and existing pixels.
        long_before = capture('Documents-Long-Before', before, long_id, long_bounds, 'WebsiteParity.md', ready=fallback_ready)
        assert long_before['ReadingProgressPixels']['FillFraction'] < .9, 'Long document has no reading remaining'
        assert '图表组件加载失败' in (long_before['Document'].get('MermaidState') or {}).get('Text', ''), 'Unsupported diagram has no explicit fallback'
        rejected = [line for line in healthy()[before:].splitlines() if line.startswith('XFE_BROWSER_SCRIPT_REJECTED|') and 'mermaid.min.js' in line and '|Reason|ResourceBudget' in line]
        assert rejected and not any(line.startswith('XFE_BROWSER_SCRIPT_BEGIN|') and 'mermaid.min.js' in line for line in healthy()[before:].splitlines()), 'Diagram budget rejection was not truthful'
        pane = viewport(long_bounds); move(pane[0] + 8, pane[1] + min(180, pane[3] // 2))
        wheel_start = len(healthy()); wheel(6)
        long_after = capture('Documents-Long-After', wheel_start, long_id, long_bounds, 'WebsiteParity.md', previous=long_before, heading=False, ready=fallback_ready)
        assert long_after['Presentation']['ScrollY'] > long_before['Presentation']['ScrollY'] + 30, 'Real wheel did not move the displayed snapshot'
        result['LongScroll'] = {'Passed': True, 'WindowId': long_id, 'PhysicalWheelSteps': 6, 'Before': long_before, 'After': long_after,
                              'NativeFillWidthIncrease': long_after['ReadingProgressPixels']['NativeFillWidth'] - long_before['ReadingProgressPixels']['NativeFillWidth']}

        # Scroll the real pane until its actual fallback region is on screen.
        fallback = None
        for steps in range(0, 41, 2):
            after = wheel_start if steps == 0 else len(healthy())
            if steps: move(pane[0] + 8, pane[1] + 180); wheel(2)
            def visible(item):
                mermaid = item.get('MermaidState') or {}
                rect = mermaid.get('Bounds') or {}
                # A toolbar alone cannot qualify the source fallback. Its
                # complete region must fit after the real rejection completed.
                return fallback_ready(item) and 'flowchart LR' in mermaid.get('Text', '') and \
                    rect.get('Y', -10000) >= 0 and rect.get('Height', 0) > 0 and rect.get('Y', 10000) + rect['Height'] < pane[3]
            try:
                item, end = document(after, 'WebsiteParity.md', visible, seconds=3 if steps < 40 else 30)
            except AssertionError:
                continue
            ack = presented(end, long_id, item)
            move(1200, 710); screenshot('Documents-Source-Fallback')
            region = item['MermaidState']; rect = region['Bounds']
            rgb = tuple(int(normalized_color(region['Style']['Color'])[index:index+2], 16) for index in (0, 2, 4))
            bounds_pixels = [pane[0] + max(0, int(rect['X'])), pane[1] + max(0, int(rect['Y'])),
                             pane[0] + min(pane[2], int(rect['X'] + rect['Width'])), pane[1] + min(pane[3], int(rect['Y'] + rect['Height']))]
            with Image.open(proof_directory / 'Documents-Source-Fallback.png') as frame:
                ink = sum(all(abs(pixel[channel] - rgb[channel]) <= 12 for channel in range(3)) for pixel in frame.convert('RGB').crop(bounds_pixels).getdata())
            assert ink >= 20, 'Original diagram source fallback has no corresponding visible glyph ink'
            result['Screenshots']['Documents-Source-Fallback.png'] = hashlib.sha256((proof_directory / 'Documents-Source-Fallback.png').read_bytes()).hexdigest()
            fallback = {'Passed': True, 'Document': item, 'Presentation': ack, 'OriginalSourceFallbackVisible': True,
                        'DiagramRenderingSupported': False, 'RejectedBeforeEvaluation': True, 'RejectionDiagnostics': rejected,
                        'MatchingGlyphPixels': ink, 'NativeBounds': bounds_pixels, 'Screenshot': 'Documents-Source-Fallback.png'}
            break
        assert fallback is not None, 'Actual scrolling never exposed the original diagram source fallback'
        result['SourceFallback'] = fallback
        close_editor(); healthy()
        result.update(Passed=True, NoRenderFailures=True)
        persist()
        return result
    except Exception as error:
        result['Failure'] = str(error); persist(); raise