"""Sustained native editor lifecycle proof through physical QMP input only.

The official Monaco engine and its timers remain active. Models are read only
through fixed guest diagnostics; previews use the original website assets and
are checked independently against actual scanout pixels. No guest JavaScript,
model, DOM or memory is injected, and no extra guest RAM is requested here.
"""
import hashlib
import json
from pathlib import Path
import re
import time

from NativeCSharpEvidence import CSHARP_PROOF
from TestWebsiteDocument import (capture_document, verify_code, wait_document,
                                 wait_document_presented_frame, RENDER_FRAMES)


MODEL = re.compile(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END', re.S)
BASIC_MODEL = re.compile(r'XFE_BASIC_EDITOR_VALUE_BEGIN\r?\nWindow\|(\d+)\|Path\|([^|\r\n]+)'
                         r'\|SourceCharacters\|(\d+)\|Truncated\|([01])\r?\n(.*?)\r?\nXFE_BASIC_EDITOR_VALUE_END', re.S)
RENDER_STATS = re.compile(r'^XFE_RENDER_STATS\|([^\r\n]+)\r?\n', re.M)
EDITOR_RETIRED = re.compile(r'^XFE_EDITOR_BACKGROUND\|Retired\|Window\|(\d+)'
                            r'\|Generation\|(\d+)\|Thread\|(\d+)\r?\n', re.M)
DESKTOP_FLAGS = re.compile(r'^XFE_DESKTOP_FLAGS\|[01]\|[01]\|Active\|\d+\r?\n', re.M)
RELEASE_PACKET = re.compile(
    r'^XFE_BROWSER_RELEASE\|HandlesBefore\|(?P<DomBefore>\d+)\|HandlesAfter\|(?P<DomAfter>\d+)\r?\n'
    r'XFE_BROWSER_WEAK_RELEASE\|TablesBefore\|(?P<TablesBefore>\d+)\|EntriesBefore\|(?P<EntriesBefore>\d+)'
    r'\|TablesAfter\|(?P<TablesAfter>\d+)\|EntriesAfter\|(?P<EntriesAfter>\d+)'
    r'\|Created\|(?P<Created>\d+)\|Disposed\|(?P<Disposed>\d+)'
    r'\|OwnerHandlesCreated\|(?P<OwnerCreated>\d+)\|OwnerHandlesDisposed\|(?P<OwnerDisposed>\d+)'
    r'\|Page\|(?P<Page>\d+)\|Reason\|(?P<Reason>Release|NavigationFailed|Replace)\r?\n', re.M)
FAILURES = ('CPU EXCEPTION:', '[FAILFAST]', '*** UNHANDLED EXCEPTION',
            'XFE_RUNTIME_OOM|', 'XFE_RUNTIME_STACK_OOM|')
DOCUMENTS = 'X:\\Users\\Default\\Documents\\'
INITIAL = {
    'Hello.cs': 'using System;\n\nConsole.ForegroundColor = ConsoleColor.Cyan;\n'
                'Console.WriteLine("Hello XFESystem!");\nConsole.ResetColor();\n',
    'Hello.js': "const greeting = 'XFESystem';\nconsole.log(`Hello from ${greeting}!`);\n",
    'Readme.md': '# XFESystem Markdown\n\n**Website renderer** is active.\n\n'
                 '- C# desktop\n- JavaScript preview\n',
}


def render_samples(log, after=0):
    """Only complete newline-terminated native records are evidence."""
    samples = []
    for match in RENDER_STATS.finditer(log, after):
        parts = match.group(1).split('|')
        if len(parts) < 3 or len(parts) % 2 != 1:
            continue
        try:
            values = {parts[index]: int(parts[index + 1]) for index in range(1, len(parts), 2)}
            values['Timestamp'] = int(parts[0])
        except ValueError:
            continue
        if not {'RawFreePages', 'RawTotalPages', 'RawLargestFreeRunPages'}.issubset(values):
            continue
        values['SerialOffset'] = match.start()
        values['SerialEnd'] = match.end()
        samples.append(values)
    return samples


def memory_observation(samples, seconds):
    assert samples, 'No complete raw-page/UI heartbeat samples'
    assert all(sample['RawTotalPages'] == samples[0]['RawTotalPages'] for sample in samples)
    assert all(a['Timestamp'] < b['Timestamp'] for a, b in zip(samples, samples[1:])), 'Native heartbeat time did not advance'
    # Compare the best free-page levels in the first and final observation
    # thirds. A native GC may reclaim dead transient objects in either window;
    # a single pre-GC low-water point alone does not prove a managed leak.
    window = max(1, len(samples) // 3)
    initial_peak = max(sample['RawFreePages'] for sample in samples[:window])
    final_peak = max(sample['RawFreePages'] for sample in samples[-window:])
    growth = max(0, initial_peak - final_peak) * 4096
    return {'Seconds': seconds, 'SampleCount': len(samples), 'Samples': samples,
            'FirstWindowPeakFreePages': initial_peak, 'FinalWindowPeakFreePages': final_peak,
            'PeakToPeakRawPageConsumptionBytes': growth,
            'MinimumRawFreePages': min(sample['RawFreePages'] for sample in samples),
            'MinimumLargestContiguousFreePages': min(sample['RawLargestFreeRunPages'] for sample in samples),
            'RawPageConsumptionBudgetBytes': 64 * 1024 * 1024,
            'WithinRawPageConsumptionBudget': growth <= 64 * 1024 * 1024,
            'Meaning': 'Bounded physical-page observation, not an unlimited leak guarantee or exact managed live-byte count.'}


def closed_cycle_handle_observation(log, actor_windows, files_id, closed_after,
                                    desktop_snapshots, warm_baseline=None):
    """Read a settled closed triplet; active pages and incomplete reports cannot qualify.

    actor_windows maps this triplet's actual CS/JS editor ids to their source
    readback offsets. A matching complete retirement must follow that readback.
    The two adjacent native reports must both follow all retirements and the
    final Markdown close, and their latest sealed desktop reports must contain
    only the active Files window. No guest input or collection is performed.
    """
    assert len(actor_windows) == 2 and all(window_id > 0 for window_id in actor_windows)
    retirements = []
    for window_id, after in actor_windows.items():
        matches = [match for match in EDITOR_RETIRED.finditer(log, after)
                   if int(match.group(1)) == window_id]
        if not matches:
            return None
        match = matches[0]
        retirements.append({'WindowId': window_id, 'Generation': int(match.group(2)),
                            'ThreadId': int(match.group(3)), 'SerialOffset': match.start(),
                            'SerialEnd': match.end(), 'CompleteMarker': match.group(0).rstrip('\r\n')})
    lower_bound = max(closed_after, *(item['SerialEnd'] for item in retirements))
    samples = render_samples(log, lower_bound)
    states = desktop_snapshots(log)

    def closed_state(sample):
        candidates = [state for state in states
                      if closed_after <= state['SerialOffset'] < sample['SerialOffset']]
        if not candidates:
            return None
        state = candidates[-1]
        # A malformed or unfinished newer header cannot silently fall back to
        # an older complete Files-only report.
        flags = list(DESKTOP_FLAGS.finditer(log, state['SerialOffset'], sample['SerialEnd']))
        if not flags or flags[-1].start() != state['SerialOffset']:
            return None
        if (state['ActiveId'] != files_id or state['StartMenuOpen'] or state['WindowSwitching'] or
            {key: value['Application'] for key, value in state['Windows'].items()} != {files_id: 'Files'} or
            state['Windows'][files_id]['Minimized']):
            return None
        return state

    for first, second in zip(samples, samples[1:]):
        allocated = first.get('DependentHandles')
        frequency = first.get('Frequency', 0)
        if (allocated is None or allocated < 0 or second.get('DependentHandles') != allocated or
            frequency <= 0 or second.get('Frequency') != frequency or
            second['Timestamp'] - first['Timestamp'] < frequency or
            len(list(RENDER_STATS.finditer(log, first['SerialOffset'], second['SerialEnd']))) != 2):
            continue
        first_state, second_state = closed_state(first), closed_state(second)
        if first_state is None or second_state is None:
            continue
        baseline = allocated if warm_baseline is None else warm_baseline
        return {'ActorRetirements': retirements, 'NativeSerialLowerBound': lower_bound,
                'FilesWindowId': files_id, 'ClosedDesktopStates': [first_state, second_state],
                'ConsecutiveNativeSamples': [first, second], 'AllocatedDependentSlots': allocated,
                'NativeQuietSeconds': (second['Timestamp'] - first['Timestamp']) / frequency,
                'WarmBaselineAllocatedDependentSlots': baseline,
                'DeltaFromWarmBaseline': allocated - baseline, 'WithinWarmBaseline': allocated <= baseline}
    return None


def require_no_closed_cycle_handle_growth(allocated, warm_baseline):
    assert allocated <= warm_baseline, ('Closed-cycle native dependent slots grew above the first complete '
                                      'CS/JS/Markdown warm baseline', allocated, warm_baseline,
                                      allocated - warm_baseline)


def closed_cycle_release_observation(log, entries, retired, end, previously_released_pages):
    """Bind atomic DOM/weak release packets to the actual close/retire intervals."""
    start = min(entry['SerialStart'] for entry in entries)
    packets = []
    for match in RELEASE_PACKET.finditer(log, start, end):
        values = {key: int(value) if key != 'Reason' else value for key, value in match.groupdict().items()}
        assert values['DomAfter'] == values['TablesAfter'] == values['EntriesAfter'] == 0
        assert values['Created'] == values['Disposed'] and values['OwnerCreated'] == values['OwnerDisposed']
        assert values['Page'] > 0
        values.update({'SerialOffset': match.start(), 'SerialEnd': match.end()})
        packets.append(values)
    dom_lines = list(re.finditer(r'^XFE_BROWSER_RELEASE\|[^\r\n]*\r?\n', log[start:end], re.M))
    weak_lines = list(re.finditer(r'^XFE_BROWSER_WEAK_RELEASE\|[^\r\n]*\r?\n', log[start:end], re.M))
    assert len(dom_lines) == len(weak_lines) == len(packets), 'Incomplete or malformed paired DOM/weak release record'
    expected_intervals = []
    retirements = {item['WindowId']: item for item in retired}
    for entry in entries:
        preview = entry['Preview']
        expected_intervals.append({'Kind': 'Preview', 'WindowId': preview['BrowserId'],
            'Start': preview['CloseSerialStart'], 'End': preview['ClosedToEditor']['SerialOffset']})
        if entry['Path'].endswith(('.cs', '.js')):
            actor = retirements[entry['EditorId']]
            expected_intervals.append({'Kind': 'EditorActor', 'WindowId': entry['EditorId'],
                'Start': entry['CloseSerialStart'], 'End': actor['SerialOffset']})
    assert len(expected_intervals) == 5
    assigned = []
    for interval in expected_intervals:
        matches = [packet for packet in packets if packet['Reason'] == 'Release' and
                   interval['Start'] <= packet['SerialOffset'] and packet['SerialEnd'] <= interval['End']]
        assert len(matches) == 1, ('No unique corresponding owner release in native close interval', interval, matches)
        assigned.append(dict(interval, Release=matches[0]))
    pages = [item['Release']['Page'] for item in assigned]
    assert len(set(pages)) == 5 and not set(pages) & previously_released_pages, 'Owner page released twice or reused from historical evidence'
    assert sum(packet['Reason'] == 'Release' for packet in packets) == 5, 'Unmatched owner release outside the corresponding close intervals'
    return {'Passed': True, 'ExpectedOwnerReleases': assigned, 'AllReleasePackets': packets,
            'ReleasedPageIds': pages, 'PairedDomAndWeakCountsChecked': True}


def closed_cycle_physical_budget(observations):
    first, latest = observations[0], observations[-1]
    baseline = max(sample['RawFreePages'] for sample in first['ConsecutiveNativeSamples'])
    current = max(sample['RawFreePages'] for sample in latest['ConsecutiveNativeSamples'])
    consumption = max(0, baseline - current) * 4096
    return {'WarmClosedBaselinePeakFreePages': baseline, 'CurrentClosedPeakFreePages': current,
            'RawPageConsumptionBytes': consumption, 'RawPageConsumptionBudgetBytes': 64 * 1024 * 1024,
            'Passed': consumption <= 64 * 1024 * 1024, 'BaselineRolled': False,
            'Meaning': 'Same Files-only closed/retired state; bounded physical-page retention check, not an exact managed live-byte walk.'}


def run_editor_stability(qmp, read_log, wait_log, screenshot, width, height, scale,
                         pointer, desktop_snapshots, idle_seconds=300, cycles=12, native_reader=None):
    assert 0 <= idle_seconds <= 3600 and 0 <= cycles <= 120
    begun = time.monotonic()
    result = {'Kind': 'NativeEditorSustainedPhysicalInputProof', 'Passed': False,
              'IdleSecondsRequested': idle_seconds, 'LifecycleCyclesRequested': cycles,
              'Input': 'QMP physical PS/2 keyboard and relative mouse events',
              'PhysicalKeyboardTiming': {'KeyDownMilliseconds': 40, 'PostReleaseWaitMilliseconds': 80,
                  'Release': 'Explicit reverse-order QMP key-up events; no HMP delayed automatic release'},
              'HostJavaScriptInjection': False, 'GuestTimersDisabled': False,
              'GpuExecutionClaimed': False, 'FullFirstPaintUnderOneSecondClaimed': False,
              'JavaScriptPreviewBehavior': 'Source highlighting using website scripts; source is not executed as a local program.',
              'Actions': [], 'Lifecycles': [], 'ShortcutSamples': [], 'RawPageSamples': [],
              'ClosedCycleHandleObservations': [], 'ClosedCycleHandleRequirementMet': False,
              'ClosedCycleOwnershipObservations': [], 'ClosedCycleOwnershipRequirementMet': False,
              'ClosedCyclePhysicalBudgetObservations': [],
              'ClosedCycleHandlePolicy': {'TimeoutSeconds': 15, 'MinimumNativeQuietSeconds': 1,
                  'WarmBaseline': 'First complete CS/JS/Markdown triplet after both editor actors retire',
                  'AllowedAllocatedSlotGrowth': 0, 'ForcedCollection': False}}
    folder = Path(screenshot('EditorStability-Initial')).resolve().parent
    output = folder / 'EditorStability.json'
    scan_offset = 0
    expected = dict(INITIAL)
    released_pages = set()
    if cycles:
        assert native_reader is not None, 'Native ownership reader is required for lifecycle qualification'
        from NativeDependentSlotEvidence import ClosedCycleOwnershipTracker
        ownership_tracker = ClosedCycleOwnershipTracker()
        result['NativeOwnershipLayout'] = native_reader.provenance

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

    def safe():
        nonlocal scan_offset
        log = read_log()
        for sample in render_samples(log, scan_offset):
            sample['HostObservedSeconds'] = time.monotonic() - begun
            result['RawPageSamples'].append(sample)
            scan_offset = sample['SerialEnd']
        assert not any(marker in log for marker in FAILURES), log[-6000:]
        return log

    def phase(name, **fields):
        result['CurrentPhase'] = name
        result['Actions'].append({'Action': name, 'HostSeconds': time.monotonic() - begun, **fields})
        safe(); persist()

    def move(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(.015)

    def click(x, y):
        move(x, y)
        result['Actions'].append({'Action': 'PhysicalClick', 'LayoutPosition': [x, y],
            'RequestedNativePointerPosition': list(pointer), 'HostSeconds': time.monotonic() - begun})
        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'}}]})

    def key(code):
        keys = code.split('-')
        qmp('input-send-event', {'events': [{'type': 'key', 'data': {'down': True,
            'key': {'type': 'qcode', 'data': value}}} for value in keys]})
        time.sleep(.04)
        qmp('input-send-event', {'events': [{'type': 'key', 'data': {'down': False,
            'key': {'type': 'qcode', 'data': value}}} for value in reversed(keys)]})
        time.sleep(.08)

    def type_text(value):
        for character in value:
            code = {'/': 'slash', ' ': 'spc', '\n': 'ret', '-': 'minus'}.get(character, character)
            assert re.fullmatch(r'[a-z0-9]|slash|spc|ret|minus', code), ('Unsupported physical test character', character)
            key(code)

    def wait_state(start, predicate, action, timeout=45):
        deadline = time.monotonic() + timeout
        last = None
        while time.monotonic() < deadline:
            log = safe()
            states = desktop_snapshots(log[start:])
            if states:
                last = states[-1]
                if predicate(last):
                    last = dict(last, SerialOffset=start + last['SerialOffset'])
                    result['Actions'].append({'Action': action, 'HostSeconds': time.monotonic() - begun, 'Acknowledgement': last})
                    persist(); return last
            time.sleep(.05)
        raise AssertionError((action, 'No complete corresponding native desktop acknowledgement', last, read_log()[-5000:]))

    def window_map(state):
        return {window_id: row['Application'] for window_id, row in state['Windows'].items()}

    def wait_model(start, source, basic=False, timeout=150, path=None, window_id=None):
        pattern = BASIC_MODEL if basic else MODEL
        deadline = time.monotonic() + timeout
        latest = None
        while time.monotonic() < deadline:
            log = safe()
            models = list(pattern.finditer(log[start:]))
            if models:
                latest = models[-1].group(5 if basic else 1).replace('\r', '')
                if latest == source:
                    if basic:
                        assert models[-1].group(2).startswith(DOCUMENTS) and models[-1].group(4) == '0'
                        assert int(models[-1].group(3)) == len(source)
                        if path is not None: assert models[-1].group(2) == path
                        if window_id is not None: assert int(models[-1].group(1)) == window_id
                    return {'Characters': len(source), 'Sha256': hashlib.sha256(source.encode('utf-8')).hexdigest(),
                            'SerialOffset': start + models[-1].start(), 'ExactSource': True,
                            'Readback': 'Actual basic editor source' if basic else 'Actual Monaco getValue'}
            time.sleep(.1)
        raise AssertionError(('Actual editor source mismatch', source, latest, basic))

    def ready_editor(name, state, start):
        editor_id = state['ActiveId']
        if name.endswith('.md'):
            return wait_model(start, expected[name], basic=True, path=DOCUMENTS + name, window_id=editor_id)
        wait_log('XFE_EDITOR_BACKGROUND|Ready|Window|' + str(editor_id) + '|', start, 150)
        proof = wait_model(start, expected[name])
        if name.endswith('.cs'):
            deadline = time.monotonic() + 60
            while time.monotonic() < deadline:
                candidates = []
                for match in CSHARP_PROOF.finditer(safe()[start:]):
                    candidate = json.loads(match.group(1))
                    if candidate.get('Language') == 'csharp' and candidate.get('ModelValue') == expected[name]:
                        candidates.append(candidate)
                if candidates:
                    actual = candidates[-1]
                    assert actual['Kind'] == 'ActualMonacoWebsiteCSharpDomReadback' and actual['HostScreenshotUsed'] is False
                    assert actual['ModelLength'] == len(expected[name]) and actual['ViewLines']
                    proof['ActualCSharpDom'] = actual
                    return proof
                time.sleep(.1)
            raise AssertionError('No actual C# Monaco DOM proof for ' + name)
        return proof

    def open_file(name):
        phase('OpenFile', Path=DOCUMENTS + name)
        states = desktop_snapshots(safe())
        assert states
        current = states[-1]
        assert current['ActiveId'] == files_id and window_map(current) == {files_id: 'Files'}
        files = current['Windows'][files_id]
        x, y, _, _ = files['Bounds']
        index = ('Hello.cs', 'Hello.js', 'Readme.md').index(name)
        start = len(safe())
        click(x + 70, y + 110 + index * 28)
        state = wait_state(start, lambda item: item['ActiveId'] != files_id and
            item['Windows'].get(item['ActiveId'], {}).get('Application') == 'Editor' and
            len(item['Windows']) == 2 and item['Windows'].get(files_id, {}).get('Application') == 'Files', 'File editor opened')
        proof = ready_editor(name, state, start)
        return state['ActiveId'], state['Windows'][state['ActiveId']]['Bounds'], proof

    def save(name):
        start = len(safe()); before = time.monotonic(); key('ctrl-s')
        marker = 'XFE_DESKTOP_SAVED|' + DOCUMENTS + name + '|' + str(len(expected[name]))
        wait_log(marker, start, 60)
        deadline = time.monotonic() + 10
        while not re.search('^' + re.escape(marker) + r'\r?\n', safe()[start:], re.M):
            assert time.monotonic() < deadline, 'No complete corresponding save acknowledgement'
            time.sleep(.05)
        safe()
        return {'Path': DOCUMENTS + name, 'Characters': len(expected[name]),
                'HostInputToSaveAcknowledgementSeconds': time.monotonic() - before, 'SerialOffset': start}

    def append(name, serial):
        phase('AppendPhysicalInput', Path=DOCUMENTS + name, Suffix=serial)
        before = len(safe())
        if name.endswith('.md'):
            # The basic editor has line Home/End rather than Monaco Ctrl+End.
            for _ in range(expected[name].count('\n') + 1): key('down')
            key('end')
            suffix = '\nstability md ' + serial
        else:
            key('ctrl-end')
            suffix = '\n//xfe' + serial
        type_text(suffix)
        expected[name] += suffix
        return wait_model(before, expected[name], basic=name.endswith('.md'), path=DOCUMENTS + name)

    def preview(name, editor_id, tag):
        phase('Preview', Path=DOCUMENTS + name)
        frames = RENDER_FRAMES.findall(safe())
        before_frames = int(frames[-1]) if frames else 0
        start = len(safe()); key('f5')
        state = wait_state(start, lambda item: len(item['Windows']) == 3 and
            item['Windows'].get(item['ActiveId'], {}).get('Application') == 'Browser' and
            item['Windows'].get(editor_id, {}).get('Application') == 'Editor' and
            item['Windows'].get(files_id, {}).get('Application') == 'Files', 'Preview browser opened', timeout=120)
        browser_id = state['ActiveId']
        if name.endswith('.md'):
            proof = wait_document(safe, start, lambda item: (item.get('Heading') or {}).get('Text') == 'XFESystem Markdown' and
                all(line in item.get('BodyText', '') for line in expected[name].split('\n') if line.startswith('stability md ')), timeout=150)
            assert 'Website renderer' in proof['BodyText'] and 'C# desktop' in proof['BodyText']
        else:
            proof = wait_document(safe, start, lambda item: any(block.get('CodeText') == expected[name]
                for block in item.get('CodeBlocks', [])), timeout=150)
            verify_code(proof, 'csharp' if name.endswith('.cs') else 'javascript', expected[name])
        # DOM diagnostics are emitted before rendering. Require a later real
        # completed frame before independently scanning its native glyph ink.
        present = {}
        wait_document_presented_frame(safe, start, proof, before_frames, present, timeout=60)
        move(width / scale - 110, height / scale - 100)
        image = capture_document(screenshot, tag + '-WebsitePreview', safe, proof, scale,
                                 require_heading=name.endswith('.md'), timeout=60)
        before = len(safe()); key('alt-f4')
        closed = wait_state(before, lambda item: item['ActiveId'] == editor_id and
            window_map(item) == {files_id: 'Files', editor_id: 'Editor'}, 'Close actual website preview and return to saved editor')
        return {'BrowserId': browser_id, 'DomReadback': proof, 'PresentedFrame': present,
                'Framebuffer': image, 'JavaScriptExecutedAsLocalProgram': False,
                'CloseSerialStart': before, 'ClosedToEditor': closed}

    def close_editor(editor_id):
        before = len(safe()); key('alt-f4')
        return wait_state(before, lambda item: item['ActiveId'] == files_id and window_map(item) == {files_id: 'Files'},
                          'Close saved editor and return to Files')

    def lifecycle(name, number, leave_open=False):
        entry = {'Path': DOCUMENTS + name, 'Iteration': number, 'Passed': False, 'SerialStart': len(safe())}
        result['Lifecycles'].append(entry); persist()
        editor_id, bounds, original = open_file(name)
        entry.update({'EditorId': editor_id, 'InitialBounds': bounds, 'ReopenedExactSource': original})
        entry['EditedExactSource'] = append(name, f'{number:02d}')
        entry['Save'] = save(name)
        entry['Preview'] = preview(name, editor_id, 'EditorStability-' + name.replace('.', '-') + '-' + str(number))
        if not leave_open:
            entry['CloseSerialStart'] = len(safe())
            entry['ClosedToFiles'] = close_editor(editor_id)
        entry['Passed'] = True; persist()
        return editor_id

    def observe_closed_cycle_handles(number):
        entries = [entry for entry in result['Lifecycles'] if entry['Iteration'] == number]
        assert len(entries) == 3 and all(entry['Passed'] for entry in entries)
        actors = {entry['EditorId']: entry['ReopenedExactSource']['SerialOffset'] for entry in entries
                  if entry['Path'].endswith(('.cs', '.js'))}
        closed_after = entries[-1]['ClosedToFiles']['SerialOffset']
        baseline = result.get('ClosedCycleWarmBaselineAllocatedDependentSlots')
        phase('ObserveClosedCycleHandles', Cycle=number)
        started = time.monotonic(); deadline = started + 15
        while time.monotonic() < deadline:
            observation = closed_cycle_handle_observation(safe(), actors, files_id, closed_after,
                                                           desktop_snapshots, baseline)
            if observation is not None:
                observation.update({'Cycle': number, 'NoInputDuringObservation': True,
                                    'ForcedCollection': False, 'HostWaitSeconds': time.monotonic() - started})
                result['ClosedCycleHandleObservations'].append(observation)
                if baseline is None:
                    baseline = observation['AllocatedDependentSlots']
                    result['ClosedCycleWarmBaselineAllocatedDependentSlots'] = baseline
                persist()
                # Preserve the original global-count result. A total increase
                # is never relabelled as passing this fixed-baseline check.
                physical = closed_cycle_physical_budget(result['ClosedCycleHandleObservations'])
                result['ClosedCyclePhysicalBudgetObservations'].append(dict(physical, Cycle=number))
                persist()
                assert physical['Passed'], 'Closed-cycle physical retention exceeded the original64MiB budget'
                release = closed_cycle_release_observation(safe(), entries, observation['ActorRetirements'],
                    observation['ConsecutiveNativeSamples'][-1]['SerialEnd'], released_pages)
                value, capture = native_reader.capture(number, observation)
                qualification = ownership_tracker.qualify(value)
                result['ClosedCycleOwnershipObservations'].append({'Cycle': number, 'Ownership': qualification,
                    'NativeCapture': capture, 'Release': release, 'GlobalCountWithinOriginalWarmBaseline': observation['WithinWarmBaseline']})
                released_pages.update(release['ReleasedPageIds'])
                persist()
                return
            time.sleep(.1)
        persist()
        raise AssertionError(('No settled Files-only pair of native dependent-slot samples after both '
                              'corresponding CS/JS editor retirements within 15 seconds', number, actors))

    def shortcut(editor_id, code, active_id, minimized):
        before = len(safe()); started = time.monotonic(); key(code)
        state = wait_state(before, lambda item: item['ActiveId'] == active_id and
            window_map(item) == {files_id: 'Files', editor_id: 'Editor'} and
            all(bool(row['Minimized']) == minimized for row in item['Windows'].values()), 'Focused idle ' + code, timeout=15)
        elapsed = time.monotonic() - started
        result['ShortcutSamples'].append({'Shortcut': code, 'ActiveId': active_id,
            'HostInputToAcknowledgementSeconds': elapsed, 'UnderOneSecond': elapsed < 1,
            'AcknowledgementSerialOffset': state['SerialOffset']})
        persist()

    try:
        phase('OpenFiles')
        start = len(safe()); click(60, 120)
        state = wait_state(start, lambda item: len(item['Windows']) == 1 and
            item['Windows'].get(item['ActiveId'], {}).get('Application') == 'Files', 'Open Documents Files window')
        files_id = state['ActiveId']
        first_editor = lifecycle('Hello.cs', 0, leave_open=True)
        phase('FocusedIdle', EditorId=first_editor, Seconds=idle_seconds)
        idle_begin = time.monotonic(); idle_after = len(safe()); last_heartbeat = idle_begin
        last_count = len(result['RawPageSamples']); next_shortcut = idle_begin + 30; next_persist = idle_begin + 5
        while time.monotonic() - idle_begin < idle_seconds:
            safe()
            if len(result['RawPageSamples']) != last_count:
                last_count = len(result['RawPageSamples']); last_heartbeat = time.monotonic()
            assert time.monotonic() - last_heartbeat < 15, 'Native UI/raw-page heartbeat stalled for 15 seconds'
            now = time.monotonic()
            if now >= next_shortcut:
                shortcut(first_editor, 'meta_l-d', 0, True)
                shortcut(first_editor, 'meta_l-d', first_editor, False)
                next_shortcut = time.monotonic() + 30
            if now >= next_persist:
                persist(); next_persist = now + 5
            time.sleep(.25)
        idle_samples = [sample for sample in result['RawPageSamples'] if sample['SerialOffset'] >= idle_after]
        observed = time.monotonic() - idle_begin
        if idle_seconds >= 10:
            assert len(idle_samples) >= max(5, int(idle_seconds / 3)), ('Insufficient continuing native heartbeat samples', len(idle_samples))
            result['FocusedIdleObservation'] = memory_observation(idle_samples, observed)
            persist()
            assert result['FocusedIdleObservation']['WithinRawPageConsumptionBudget'], 'Focused editor exceeded the stated 64 MiB raw-page growth budget'
        else:
            result['FocusedIdleObservation'] = {'Seconds': observed, 'Samples': idle_samples, 'LongIdleAccepted': False,
                'Meaning': 'Short harness exercise only; does not satisfy the default 300-second idle acceptance.'}
        close_editor(first_editor)
        if cycles:
            deadline = time.monotonic() + 15
            while not any(int(marker.group(1)) == first_editor for marker in EDITOR_RETIRED.finditer(safe())):
                assert time.monotonic() < deadline, 'The idle anchor editor did not retire before lifecycle ownership observation'
                time.sleep(.1)
        for number in range(1, cycles + 1):
            for name in ('Hello.cs', 'Hello.js', 'Readme.md'):
                lifecycle(name, number)
            observe_closed_cycle_handles(number)
        # Reopen once more to prove the final saved revisions survived every
        # intervening close, rather than trusting only a same-length save ACK.
        result['FinalReopenChecks'] = []
        for name in (('Hello.cs', 'Hello.js', 'Readme.md') if cycles else ('Hello.cs',)):
            editor_id, _, proof = open_file(name)
            result['FinalReopenChecks'].append({'Path': DOCUMENTS + name, 'ExactSource': proof})
            close_editor(editor_id)
        phase('Completed')
        result['Passed'] = True
        result['LongIdleRequirementMet'] = idle_seconds >= 300 and observed >= 300
        result['FullLifecycleRequirementMet'] = cycles >= 12
        result['ClosedCycleHandleRequirementMet'] = cycles >= 12 and len(result['ClosedCycleHandleObservations']) == cycles and all(
            item['WithinWarmBaseline'] for item in result['ClosedCycleHandleObservations'])
        result['ClosedCycleOwnershipRequirementMet'] = cycles >= 12 and len(result['ClosedCycleOwnershipObservations']) == cycles
        result['FinalSavedSources'] = {DOCUMENTS + name: {'Characters': len(value),
            'Sha256': hashlib.sha256(value.encode('utf-8')).hexdigest()} for name, value in expected.items()}
        return result
    except Exception as error:
        result['Failure'] = str(error)
        result['FailurePhase'] = result.get('CurrentPhase')
        result['FailureSerialTail'] = read_log()[-6000:]
        result['FailureRequestedNativePointerPosition'] = list(pointer)
        try:
            failure_frame = Path(screenshot('EditorStability-Failure')).resolve()
            result['FailureFramebuffer'] = {'Path': str(failure_frame),
                'Sha256': hashlib.sha256(failure_frame.read_bytes()).hexdigest()}
            result['FailureGuestStatus'] = qmp('query-status')
        except Exception as diagnostic_error:
            result['FailureFramebufferDiagnosticError'] = str(diagnostic_error)
        raise
    finally:
        persist()