XFESystem

"""Offline guards for settled native lifecycle evidence; no guest execution claimed."""
import ast
from pathlib import Path
import re
import unittest

from TestEditorStability import (closed_cycle_handle_observation,
                                require_no_closed_cycle_handle_growth)


# TestBoot is an executable boot harness. Reuse its exact pure report parser
# without importing its command-line/build/QEMU side effects.
tree = ast.parse((Path(__file__).parent / 'CSharpBoot/TestBoot.py').read_text(encoding='utf-8'))
parser = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == 'desktop_snapshots')
namespace = {'re': re}
exec(compile(ast.Module(body=[parser], type_ignores=[]), '<actual desktop_snapshots>', 'exec'), namespace)
desktop_snapshots = namespace['desktop_snapshots']


def retired(window_id):
    return f'XFE_EDITOR_BACKGROUND|Retired|Window|{window_id}|Generation|2|Thread|5\n'


def state(active=1, extra_window=False, minimized=0):
    result = (f'XFE_DESKTOP_FLAGS|0|0|Active|{active}\n'
              'XFE_DESKTOP_APPEARANCE|Dark|657DE2|Persistent|0\n'
              f'XFE_WINDOW_STATE|1|Files|150|66|800|500|{minimized}|0|0\n')
    if extra_window:
        result += 'XFE_WINDOW_STATE|9|Editor|172|88|800|500|0|0|0\n'
    return result


def stats(timestamp, handles, frequency=100):
    return (f'XFE_RENDER_STATS|{timestamp}|RawFreePages|1000|RawTotalPages|2000'
            f'|RawLargestFreeRunPages|40|Frequency|{frequency}|DependentHandles|{handles}\n')


class ClosedCycleHandleGuards(unittest.TestCase):
    def observe(self, log, baseline=None, actors=None, closed_after=None):
        return closed_cycle_handle_observation(log, actors or {3: 0, 5: 0}, 1,
            log.index('XFE_DESKTOP_FLAGS') if closed_after is None else closed_after,
            desktop_snapshots, baseline)

    def prefix(self):
        return retired(3) + retired(5) + state()

    def test_complete_retired_actors_and_two_stable_files_only_reports_qualify(self):
        proof = self.observe(self.prefix() + stats(100, 82) + stats(200, 82))
        self.assertEqual([3, 5], [row['WindowId'] for row in proof['ActorRetirements']])
        self.assertEqual(82, proof['AllocatedDependentSlots'])
        self.assertEqual(82, proof['WarmBaselineAllocatedDependentSlots'])
        self.assertEqual(1, proof['NativeQuietSeconds'])
        self.assertTrue(proof['WithinWarmBaseline'])
        self.assertEqual(2, len(proof['ConsecutiveNativeSamples']))

    def test_partial_or_wrong_actor_retirement_never_qualifies(self):
        for bad in ('', retired(50), retired(5).rstrip('\n')):
            with self.subTest(retirement=bad):
                # A partial last line remains unsealed before any later marker.
                suffix = ('\n' if bad and not bad.endswith('\n') else '')
                if suffix:
                    log = retired(3) + state() + stats(100, 82) + stats(200, 82) + bad
                else:
                    log = retired(3) + bad + state() + stats(100, 82) + stats(200, 82)
                self.assertIsNone(self.observe(log))

    def test_retirement_before_corresponding_model_readback_is_stale(self):
        prefix = self.prefix()
        log = prefix + stats(100, 82) + stats(200, 82)
        self.assertIsNone(self.observe(log, actors={3: len(retired(3)), 5: 0}))

    def test_both_samples_must_follow_late_actor_retirement(self):
        log = retired(3) + state() + stats(100, 82) + stats(200, 82) + retired(5) + stats(300, 82)
        self.assertIsNone(self.observe(log))
        proof = self.observe(log + stats(400, 82))
        self.assertEqual([300, 400], [row['Timestamp'] for row in proof['ConsecutiveNativeSamples']])

    def test_complete_markers_and_two_distinct_native_heartbeats_required(self):
        for samples in (stats(100, 82), stats(100, 82) + stats(200, 82).rstrip('\n'),
                        stats(100, 82) + stats(100, 82), stats(100, 82) + stats(199, 82),
                        stats(100, 82) + stats(200, 82, frequency=101)):
            with self.subTest(samples=samples):
                self.assertIsNone(self.observe(self.prefix() + samples))

    def test_equal_nonconsecutive_values_cannot_hide_a_changed_report(self):
        log = self.prefix() + stats(100, 82) + stats(200, 83) + stats(300, 82)
        self.assertIsNone(self.observe(log))
        proof = self.observe(log + stats(400, 82))
        self.assertEqual([300, 400], [row['Timestamp'] for row in proof['ConsecutiveNativeSamples']])

    def test_malformed_complete_stats_between_equal_values_breaks_consecutive_pair(self):
        damaged = stats(200, 82).replace('RawTotalPages|2000', 'RawTotalPages|bad')
        log = self.prefix() + stats(100, 82) + damaged + stats(300, 82)
        self.assertIsNone(self.observe(log))
        proof = self.observe(log + stats(400, 82))
        self.assertEqual([300, 400], [row['Timestamp'] for row in proof['ConsecutiveNativeSamples']])

    def test_active_editor_minimized_files_or_open_new_window_prevents_qualification(self):
        for bad_state in (state(active=9, extra_window=True), state(extra_window=True), state(minimized=1)):
            with self.subTest(state=bad_state):
                self.assertIsNone(self.observe(retired(3) + retired(5) + bad_state + stats(100, 82) + stats(200, 82)))
                self.assertIsNone(self.observe(self.prefix() + stats(100, 82) + bad_state + stats(200, 82)))

    def test_malformed_newer_state_cannot_reuse_an_older_closed_report(self):
        log = self.prefix() + stats(100, 82) + 'XFE_DESKTOP_FLAGS|0|0|Active|9\nBAD\n' + stats(200, 82)
        self.assertIsNone(self.observe(log))

    def test_old_files_state_before_final_markdown_close_does_not_count(self):
        log = self.prefix() + stats(100, 82) + stats(200, 82)
        self.assertIsNone(self.observe(log, closed_after=log.index('XFE_RENDER_STATS')))

    def test_growth_is_recordable_but_fails_the_fixed_warm_baseline(self):
        proof = self.observe(self.prefix() + stats(100, 83) + stats(200, 83), baseline=82)
        self.assertFalse(proof['WithinWarmBaseline'])
        self.assertEqual(1, proof['DeltaFromWarmBaseline'])
        with self.assertRaisesRegex(AssertionError, 'grew above'):
            require_no_closed_cycle_handle_growth(proof['AllocatedDependentSlots'], 82)
        for value in (82, 81):
            require_no_closed_cycle_handle_growth(value, 82)

    def test_actual_historical_aligned_slot_counts_reject_per_cycle_leak(self):
        # Captured native Files-only/Retired/DOM HandlesAfter=0 values from
        # Proof-1792001863948928300 lines38314/45882/52227/59056. The old
        # harness did not wait for a stable pair, so these test only the
        # growth gate; no new two-heartbeat native pass is claimed.
        baseline, *later = (1188, 1630, 2071, 2513)
        for value in later:
            with self.subTest(allocated=value), self.assertRaises(AssertionError):
                require_no_closed_cycle_handle_growth(value, baseline)


if __name__ == '__main__':
    unittest.main()