XFESystem

"""Bounded read-only native ownership evidence for the supported frozen runtime.

Raw dependent-handle totals remain visible. Only exact live ArrayPool TLS roots
with a same-T singleton CWT native-slot match can be classified as runtime-owned.
This reader is used exclusively by the sustained lifecycle profile, never by
first-paint or responsiveness performance measurements.
"""
import hashlib
import json
from pathlib import Path
import re
import subprocess
import time

MAX_SEGMENTS = 128
MAX_WORDS_PER_READ = 4096
MAX_TLS_ARRAY_LENGTH = 4096
MAX_THREADS = 256
MAX_READ_COMMANDS_PER_PASS = 4096
U64_MAX = (1 << 64) - 1
PINNED_CORELIB_SHA256 = '613426f7e28a74de30be95bc56dcba8991286ed396379a585ced38dbd57e493e'
PINNED_COSMOS_ORIGINAL_SHA256 = '5d63cc0687931c69f0cb24740f02172e86b9e7664074d2f7b08fcf1a405c062e'


def signed32(value):
    value &= 0xffffffff
    return value - (1 << 32) if value & (1 << 31) else value


def checked_address(address, count=1):
    assert isinstance(address, int) and isinstance(count, int)
    assert 1 <= count <= MAX_WORDS_PER_READ
    assert 0xffff800000000000 <= address <= U64_MAX - count * 8 + 1
    assert address % 8 == 0
    return address


def parse_words(text, address, count):
    checked_address(address, count)
    words = {}
    for line in text.splitlines():
        row = re.match(r'^\s*(?:0x)?([0-9a-fA-F]{1,16}):\s*(.*)$', line)
        if row is None:
            continue
        first = int(row[1], 16)
        for index, value in enumerate(re.findall(r'0x([0-9a-fA-F]{1,16})\b', row[2])):
            at = first + index * 8
            if address <= at < address + count * 8 and (at - address) % 8 == 0:
                parsed = int(value, 16)
                assert at not in words or words[at] == parsed, 'Conflicting repeated HMP word'
                words[at] = parsed
    assert len(words) == count, ('Incomplete bounded HMP read', hex(address), count, len(words), text[-200:])
    return [words[address + index * 8] for index in range(count)]


def decode_segment(address, words):
    assert address % 4096 == 0 and len(words) == 512
    slots = []
    unexpected = []
    for index in range(170):
        target, dependent, tag = words[2 + index * 3:5 + index * 3]
        kind = tag & 0xffffffff
        if kind == 6:
            slots.append({'SlotAddress': address + 16 + index * 24, 'Target': target,
                          'Dependent': dependent, 'Type': kind, 'Padding': tag >> 32})
        elif kind != 0xffffffff:
            unexpected.append({'SlotIndex': index, 'Type': kind})
    # The actual Pack/Unpack implementation puts Alive in the low sixteen bits.
    # Its source comment incorrectly describes these two fields in reverse.
    return {'Address': address, 'Next': words[0], 'FreeHeadPacked': words[1],
            'Alive': words[1] & 0xffff, 'FreeListHead': (words[1] >> 16) & 0xffff,
            'Version': words[1] >> 32, 'Slots': slots, 'UnexpectedTypes': unexpected,
            'AliveMatchesTypedSlots': (words[1] & 0xffff) == len(slots)}


class _SnapshotRead:
    def __init__(self, qmp, folder, pass_name, symbols, layout):
        self.qmp = qmp
        self.layout = layout
        self.folder = folder
        self.pass_name = pass_name
        self.symbols = symbols
        self.reads = []
        self.errors = []
        self.headers = {}
        self.object_words = {}
        self.word_cache = {}
        self.start = time.monotonic()

    def read(self, address, count=1):
        checked_address(address, count)
        assert len(self.reads) < MAX_READ_COMMANDS_PER_PASS
        command = f'x/{count}gx {address:#x}'
        assert re.fullmatch(r'x/[1-9][0-9]*gx 0x[0-9a-f]+', command)
        dump = self.qmp('human-monitor-command', {'command-line': command})
        assert isinstance(dump, str)
        self.reads.append({'Command': command, 'Address': address, 'Count': count, 'Raw': dump})
        words = parse_words(dump, address, count)
        self.word_cache.update({address + index * 8: value for index, value in enumerate(words)})
        return words

    def cached(self, address, count=1):
        checked_address(address, count)
        if all(address + index * 8 in self.word_cache for index in range(count)):
            return [self.word_cache[address + index * 8] for index in range(count)]
        return self.read(address, count)

    def many_cached(self, address, count):
        assert 0 <= count <= MAX_TLS_ARRAY_LENGTH * 4
        result = []
        while count:
            take = min(count, MAX_WORDS_PER_READ)
            result.extend(self.cached(address, take))
            address += take * 8
            count -= take
        return result

    def batch_words(self, addresses):
        pages = {}
        for address in set(addresses):
            checked_address(address)
            if address not in self.word_cache:
                pages.setdefault(address & ~4095, []).append(address)
        for addresses in pages.values():
            first, last = min(addresses), max(addresses)
            self.cached(first, (last - first) // 8 + 1)

    def array(self, address, stride_words=1, maximum=MAX_TLS_ARRAY_LENGTH, used=None):
        checked_address(address, 2)
        header = self.cached(address, 2)
        length = header[1] & 0xffffffff
        assert 0 <= length <= maximum, ('Bounded managed array length', hex(address), length, maximum)
        count = length if used is None else used
        assert 0 <= count <= length
        return {'Address': address, 'HeaderWords': header, 'Length': length, 'Used': count,
                'StrideWords': stride_words, 'DataWords': self.many_cached(address + 16, count * stride_words)}

    def runtime_pool_snapshot(self, slots):
        descriptors, pool_cells = {}, {}
        for address, names in self.symbols.items():
            for name in names:
                prefix = '__TypeThreadStaticIndexS_P_CoreLib_System_Buffers_SharedArrayPool_1<'
                pool_prefix = '__GCSTATICSS_P_CoreLib_System_Buffers_ArrayPool_1<'
                if name.startswith(prefix) and name.endswith('>'):
                    descriptors[name[len(prefix):-1]] = (address, name)
                if name.startswith(pool_prefix) and name.endswith('>'):
                    pool_cells[name[len(pool_prefix):-1]] = (address, name)
        assert descriptors and set(descriptors) == set(pool_cells) and len(descriptors) <= 256
        self.batch_words([address + offset for address, _ in descriptors.values() for offset in (0, 8)] +
                         [address for address, _ in pool_cells.values()] +
                         [self.layout['SchedulerGcCell'], self.layout['RuntimeThreadGcCell']])
        descriptor_rows = []
        managers = []
        for generic, (address, name) in sorted(descriptors.items()):
            words = self.cached(address, 2)
            row = {'ClosedT': generic, 'Descriptor': address, 'DescriptorSymbol': name,
                   'DescriptorWords': words, 'Manager': words[0],
                   'TypeIndex': signed32(words[1] & 0xffffffff),
                   'PoolGcCell': pool_cells[generic][0], 'PoolGcCellSymbol': pool_cells[generic][1]}
            descriptor_rows.append(row)
            if row['Manager'] and row['TypeIndex'] >= 0:
                managers.append(row['Manager'] + 8)
        self.batch_words(managers)
        for row in descriptor_rows:
            row['ModuleIndex'] = signed32(self.cached(row['Manager'] + 8)[0] & 0xffffffff) if row['Manager'] and row['TypeIndex'] >= 0 else None

        scheduler_base = self.cached(self.layout['SchedulerGcCell'])[0]
        assert scheduler_base
        registry = self.cached(scheduler_base + 0x18)[0]
        registry_array = self.array(registry, maximum=MAX_THREADS)
        threads = []
        self.batch_words([thread + offset for thread in registry_array['DataWords'] if thread for offset in (0x10, 0x78)])
        for index, thread in enumerate(registry_array['DataWords']):
            if thread:
                threads.append({'RegistryIndex': index, 'ThreadAddress': thread,
                                'ThreadId': signed32(self.cached(thread + 0x78)[0] & 0xffffffff),
                                'TlsRoot': self.cached(thread + 0x10)[0]})
        runtime_base = self.cached(self.layout['RuntimeThreadGcCell'])[0]
        bootstrap = self.cached(runtime_base + 8)[0] if runtime_base else 0
        thread_roots = [dict(thread, RootKind='SchedulerThread') for thread in threads]
        thread_roots.append({'ThreadId': None, 'RootKind': 'BootstrapFallback', 'TlsRoot': bootstrap})
        root_snapshots, owners = [], []
        for thread in thread_roots:
            item = dict(thread)
            item['PoolFields'] = []
            if thread['TlsRoot']:
                modules = self.array(thread['TlsRoot'])
                item['ModuleStorage'] = modules
                rows = {}
                for descriptor in descriptor_rows:
                    module, index = descriptor['ModuleIndex'], descriptor['TypeIndex']
                    if module is None or module < 0 or index < 0 or module >= modules['Length']:
                        continue
                    row_address = modules['DataWords'][module]
                    if not row_address:
                        continue
                    if module not in rows:
                        rows[module] = self.array(row_address)
                    row = rows[module]
                    if index >= row['Length'] or not row['DataWords'][index]:
                        continue
                    type_base = row['DataWords'][index]
                    type_words = self.cached(type_base, 2)
                    target = type_words[1]
                    field = {'ClosedT': descriptor['ClosedT'], 'ModuleIndex': module, 'TypeIndex': index,
                             'TypeBase': type_base, 'TypeBaseHeaderAndField': type_words,
                             'TlsFieldAddress': type_base + 8, 'Target': target}
                    item['PoolFields'].append(field)
                    if target:
                        owners.append(dict(field, ThreadId=thread['ThreadId'], RootKind=thread['RootKind'],
                            ThreadAddress=thread.get('ThreadAddress'), RegistryIndex=thread.get('RegistryIndex')))
                item['TypeStorageRows'] = rows
            root_snapshots.append(item)

        slot_map = {slot['SlotAddress']: slot for slot in slots}
        pool_registries = []
        reverse = []
        for descriptor in descriptor_rows:
            item = {'ClosedT': descriptor['ClosedT'], 'GcCell': descriptor['PoolGcCell'],
                    'GcCellSymbol': descriptor['PoolGcCellSymbol']}
            blob = self.cached(item['GcCell'])[0]
            item['GcStaticBlob'] = blob
            if blob:
                singleton = self.cached(blob + 8)[0]
                item['PoolSingleton'] = singleton
                if singleton:
                    cwt = self.cached(singleton + 8)[0]
                    item['AllTlsBucketsCwt'] = cwt
                    if cwt:
                        container = self.cached(cwt + 0x10)[0]
                        item['CwtContainer'] = container
                        if container:
                            entries = self.cached(container + 0x18)[0]
                            first_free = signed32(self.cached(container + 0x28)[0] & 0xffffffff)
                            item['FirstFreeEntry'] = first_free
                            assert first_free >= 0
                            array = self.array(entries, stride_words=2, used=first_free)
                            item['CwtEntries'] = array
                            decoded = []
                            for index in range(first_free):
                                packed, handle = array['DataWords'][index * 2:index * 2 + 2]
                                entry = {'Index': index, 'HashCode': signed32(packed & 0xffffffff),
                                         'Next': signed32(packed >> 32), 'DependentHandle': handle}
                                if handle in slot_map:
                                    slot = slot_map[handle]
                                    entry.update({'SlotTarget': slot['Target'], 'SlotDependent': slot['Dependent']})
                                    reverse.append(dict(entry, ClosedT=descriptor['ClosedT']))
                                decoded.append(entry)
                            item['DecodedUsedEntries'] = decoded
            pool_registries.append(item)

        tls_slots = [slot for slot in slots if slot['TargetObject'].get('EEType') == self.layout['TlsArrayEEType']]
        bodies = []
        for slot in tls_slots:
            body = self.array(slot['Target'], stride_words=2, maximum=27)
            assert body['Length'] == 27
            buckets = [{'Index': index, 'PooledArray': body['DataWords'][index * 2],
                        'TimestampAndPadding': body['DataWords'][index * 2 + 1]} for index in range(27)]
            self.preload_object_headers([bucket['PooledArray'] for bucket in buckets])
            for bucket in buckets:
                bucket['PooledArrayObject'] = self.describe_object(bucket['PooledArray'])
            bodies.append({'SlotAddress': slot['SlotAddress'], 'TlsArray': slot['Target'], 'Body': body,
                           'Buckets': buckets, 'ThreadOwners': [owner for owner in owners if owner['Target'] == slot['Target']],
                           'PoolRegistryOwners': [owner for owner in reverse if owner['DependentHandle'] == slot['SlotAddress']]})
        return {'DescriptorCount': len(descriptor_rows), 'Descriptors': descriptor_rows,
                'SchedulerBase': scheduler_base, 'SchedulerRegistry': registry_array,
                'RegisteredThreads': threads, 'RuntimeThreadBase': runtime_base,
                'BootstrapFallbackStorage': bootstrap, 'ThreadRootSnapshots': root_snapshots,
                'PoolRegistries': pool_registries, 'TlsArrayBodies': bodies}

    def preload_object_headers(self, addresses):
        pages = {}
        for address in set(addresses):
            if not address or address in self.object_words:
                continue
            try:
                checked_address(address)
                pages.setdefault(address & ~4095, []).append(address)
            except AssertionError:
                pass  # describe_object records the bounded failure later.
        for page, objects in sorted(pages.items()):
            first, last = min(objects), max(objects)
            # Batch only the interval within one mapped object page, never
            # speculative intervening pages; at most512words/4096bytes.
            try:
                words = self.read(first, (last - first) // 8 + 1)
                for address in objects:
                    self.object_words[address] = words[(address - first) // 8]
            except Exception:
                pass  # An individual bounded retry is recorded below.

    def describe_object(self, address):
        if address == 0:
            return {'ObjectAddress': 0, 'Null': True}
        if address in self.headers:
            return self.headers[address]
        try:
            word = self.object_words[address] if address in self.object_words else self.read(address)[0]
            mt = word & ~1
            result = {'ObjectAddress': address, 'HeaderWord': word, 'EEType': mt,
                      'EETypeSymbols': self.symbols.get(mt, []), 'ResolvedExactSymbol': mt in self.symbols}
        except Exception as error:
            result = {'ObjectAddress': address, 'ReadError': str(error)}
            self.errors.append(result)
        self.headers[address] = result
        return result

    def run(self):
        before = self.read(self.layout['DependentStore'], 3)
        head, tail, committed = before
        assert head and tail and committed % 4096 == 0
        assert 4096 <= committed <= MAX_SEGMENTS * 4096
        segments, seen, current = [], set(), head
        while current:
            assert len(segments) < MAX_SEGMENTS and current not in seen, 'Overlong or cyclic depstore chain'
            checked_address(current, 512)
            assert current % 4096 == 0
            seen.add(current)
            segment = decode_segment(current, self.read(current, 512))
            segment['RawWords'] = self.cached(current, 512)
            segments.append(segment)
            current = segment['Next']
        assert segments[-1]['Address'] == tail and len(segments) * 4096 == committed
        slots = [slot for segment in segments for slot in segment['Slots']]
        self.preload_object_headers([address for slot in slots for address in (slot['Target'], slot['Dependent'])])
        for slot in slots:
            slot['TargetObject'] = self.describe_object(slot['Target'])
            slot['DependentObject'] = self.describe_object(slot['Dependent'])
            slot['ClearedTombstone'] = slot['Target'] == 0 and slot['Dependent'] == 0
        runtime = self.runtime_pool_snapshot(slots)
        after = self.read(self.layout['DependentStore'], 3)
        return {'ReadOnly': True, 'StoreBefore': before, 'StoreAfter': after,
                'SupportedTlsArrayEEType': self.layout['TlsArrayEEType'],
                'StoreStableWithinPass': before == after, 'Segments': segments,
                'AllocatedTypedSlots': len(slots), 'AliveCounterSum': sum(segment['Alive'] for segment in segments),
                'AliveCountersMatchTypedSlots': all(segment['AliveMatchesTypedSlots'] and not segment['UnexpectedTypes'] for segment in segments),
                'Slots': slots, 'RuntimeTlsPools': runtime, 'ObjectHeaderReads': self.headers,
                'ObjectReadErrors': self.errors, 'ReadCommandCount': len(self.reads),
                'HostElapsedSeconds': time.monotonic() - self.start}

    def save_reads(self):
        target = self.folder / (self.pass_name + '-Raw.txt')
        target.write_text(''.join(f"READ {index + 1}: {item['Command']}\n{item['Raw']}\n"
                                  for index, item in enumerate(self.reads)), encoding='utf-8')
        return {'Path': str(target), 'Sha256': hashlib.sha256(target.read_bytes()).hexdigest(),
                'ReadCommandCount': len(self.reads), 'MaxWordsInAnyRead': max((item['Count'] for item in self.reads), default=0)}


def parse_symbols(text):
    symbols = {}
    for line in text.splitlines():
        match = re.match(r'^([0-9a-fA-F]{16})\s+[A-Za-z]\s+(.+)$', line)
        if match:
            symbols.setdefault(int(match[1], 16), []).append(match[2])
    return symbols


def symbol_address(symbols, name):
    matches = [address for address, names in symbols.items() if name in names]
    assert len(matches) == 1, ('Missing or ambiguous supported native symbol', name)
    return matches[0]


def require_layout_shapes(disassembly, gc_static):
    """Reject unknown machine-code field shapes, never infer a new ABI layout."""
    checks = {
        'DependentStoreB0': bool(re.search(r'addq\s+\$0xb0,', disassembly['Count'])) and
            'GCHandleSegmentStore__get_Count>' in disassembly['Count'] and
            bool(re.search(r'#\s+0x' + format(gc_static, 'x') + r'\b', disassembly['Count'])),
        'PoolTlsField8': '__GetThreadStaticBase_' in disassembly['Pool'] and
            bool(re.search(r'leaq\s+0x8\(%\w+\)', disassembly['Pool'])) and
            bool(re.search(r'movq\s+0x8\(%\w+\)', disassembly['Pool'])) and
            'ConditionalWeakTable_2<System___Canon__System___Canon>__Add>' in disassembly['Pool'],
        'CwtContainer10': bool(re.search(r'0x10\(%\w+\)', disassembly['CwtAdd'])),
        'CwtEntries18Count28Stride16Handle8': all(token in disassembly['CwtEntry']
            for token in ('0x18(', '0x28(', 'shlq\t$0x4', 'leaq\t0x10(', 'RhHandleAllocDependent>')) and
            bool(re.search(r'movq\s+%\w+, 0x8\(%\w+\)', disassembly['CwtEntry'])),
        'ThreadTls10': bool(re.search(r'movq\s+0x10\(%\w+\), %\w+', disassembly['ThreadTls'])),
        'ThreadId78': bool(re.search(r'movl\s+0x78\(%\w+\), %\w+', disassembly['ThreadId'])),
        'SchedulerRegistry18': '__GCSTATICSCosmos_Kernel_Core_Cosmos_Kernel_Core_Scheduler_SchedulerManager>' in disassembly['Registry'] and
            bool(re.search(r'movq\s+0x18\(%\w+\)', disassembly['Registry'])),
    }
    assert all(checks.values()), ('Unsupported native ownership field layout', checks)
    return checks


def ownership_fingerprint(value):
    return {'Store': value['StoreBefore'], 'Segments': [segment['RawWords'] for segment in value['Segments']],
            'RuntimeTlsPools': value['RuntimeTlsPools'],
            'ObjectHeaders': {int(key): item.get('HeaderWord') for key, item in value['ObjectHeaderReads'].items()}}


class NativeDependentSlotReader:
    """Supported-runtime reader; every guest command is bounded read-only HMP x/."""
    def __init__(self, qmp, elf, expected_kernel_sha, folder, corelib, gc_patch_record,
                 llvm_directory=None):
        self.qmp, self.elf, self.folder = qmp, Path(elf), Path(folder)
        self.folder.mkdir(parents=True, exist_ok=True)
        assert hashlib.sha256(self.elf.read_bytes()).hexdigest() == expected_kernel_sha
        assert hashlib.sha256(Path(corelib).read_bytes()).hexdigest() == PINNED_CORELIB_SHA256, 'Unsupported CoreLib10.0.12 image'
        patch_path = Path(gc_patch_record['Path'])
        patch_raw = patch_path.read_bytes()
        assert hashlib.sha256(patch_raw).hexdigest() == gc_patch_record['Sha256']
        patch = json.loads(patch_raw.decode('utf-8-sig'))
        assert patch['PinnedVersion'] == '3.0.94' and patch['OriginalPinnedSha256'] == PINNED_COSMOS_ORIGINAL_SHA256
        assert patch['AfterSha256'] == gc_patch_record['RuntimeSha256']
        assert patch['DependentHandleCountReadOnlyExport'] is True and patch['GlobalPackageCacheModified'] is False
        llvm = Path(llvm_directory) if llvm_directory else Path(__file__).resolve().parents[1] / '.tools/llvm/bin'
        nm = subprocess.run([str(llvm / 'llvm-nm.exe'), '--defined-only', '--numeric-sort', str(self.elf)],
                            check=True, capture_output=True, text=True, encoding='utf-8', timeout=30).stdout
        self.symbols = parse_symbols(nm)
        gc_static = symbol_address(self.symbols, '__NONGCSTATICSCosmos_Kernel_Core_Cosmos_Kernel_Core_Memory_GarbageCollector_GarbageCollector')
        self.layout = {'DependentStore': gc_static + 0xb0,
            'SchedulerGcCell': symbol_address(self.symbols, '__GCSTATICSCosmos_Kernel_Core_Cosmos_Kernel_Core_Scheduler_SchedulerManager'),
            'RuntimeThreadGcCell': symbol_address(self.symbols, '__GCSTATICSCosmos_Kernel_Core_Cosmos_Kernel_Core_Runtime_Thread'),
            'TlsArrayEEType': symbol_address(self.symbols, '_ZTV67__Array<S_P_CoreLib_System_Buffers_SharedArrayPoolThreadLocalArray>')}
        methods = {
            'Count': 'XfeRuntimeDependentHandleCount',
            'Pool': 'S_P_CoreLib_System_Buffers_SharedArrayPool_1<Char>__InitializeTlsBucketsAndTrimming',
            'CwtAdd': 'S_P_CoreLib_System_Runtime_CompilerServices_ConditionalWeakTable_2<System___Canon__System___Canon>__Add',
            'CwtEntry': 'S_P_CoreLib_System_Runtime_CompilerServices_ConditionalWeakTable_2_Container<System___Canon__System___Canon>__CreateEntryNoResize',
            'ThreadTls': 'RhGetThreadStaticStorage', 'ThreadId': 'XfeRuntimeCurrentSchedulerThreadId',
            'Registry': 'Cosmos_Kernel_Core_Cosmos_Kernel_Core_Scheduler_SchedulerManager__RegisterThread'}
        disassembly = {}
        for label, name in methods.items():
            symbol_address(self.symbols, name)
            disassembly[label] = subprocess.run([str(llvm / 'llvm-objdump.exe'), '--no-show-raw-insn', '-d',
                '--disassemble-symbols=' + name, str(self.elf)], check=True, capture_output=True,
                text=True, encoding='utf-8', timeout=30).stdout
        checks = require_layout_shapes(disassembly, gc_static)
        dump = self.folder / 'SupportedLayoutDisassembly.txt'
        dump.write_text('\n'.join(disassembly.values()), encoding='utf-8')
        self.provenance = {'Kind': 'Cosmos3.0.94-CoreLib10.0.12-NativeOwnershipLayoutV1',
            'KernelSha256': expected_kernel_sha, 'CoreLibSha256': PINNED_CORELIB_SHA256,
            'GcPatchReceiptSha256': gc_patch_record['Sha256'], 'GcRuntimeSha256': patch['AfterSha256'],
            'SymbolTableSha256': hashlib.sha256(nm.encode()).hexdigest(), 'ResolvedLayout': self.layout,
            'MachineCodeShapeChecks': checks, 'DisassemblyPath': str(dump.resolve()),
            'DisassemblySha256': hashlib.sha256(dump.read_bytes()).hexdigest(),
            'OnlyGuestCommandsAdded': 'human-monitor-command x/<1..4096>gx <canonical aligned address>',
            'MaxSegments': MAX_SEGMENTS, 'MaxThreads': MAX_THREADS, 'MaxArrayLength': MAX_TLS_ARRAY_LENGTH,
            'VmPaused': False, 'ForcedGc': False, 'InjectedCodeOrGuestWrites': False,
            'PerformanceEvidenceClaimed': False}
        (self.folder / 'LayoutProvenance.json').write_text(json.dumps(self.provenance, indent=2), encoding='utf-8')

    def capture(self, cycle, observation):
        folder = self.folder / f'Closed-{cycle:02d}'
        folder.mkdir()
        passes, files = [], []
        for name in ('A', 'B'):
            reader = _SnapshotRead(self.qmp, folder, name, self.symbols, self.layout)
            try:
                value = reader.run()
                target = folder / (name + '.json')
                target.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding='utf-8')
                files.append({'Path': str(target.resolve()), 'Sha256': hashlib.sha256(target.read_bytes()).hexdigest()})
                passes.append(value)
            finally:
                raw = reader.save_reads()
                files.append(raw)
        first, second = passes
        stable = ownership_fingerprint(first) == ownership_fingerprint(second)
        valid = stable and all(value['StoreStableWithinPass'] and value['AliveCountersMatchTypedSlots'] and
            not value['ObjectReadErrors'] and value['AllocatedTypedSlots'] == observation['AllocatedDependentSlots'] and
            value['AliveCounterSum'] == observation['AllocatedDependentSlots'] for value in passes)
        record = {'Cycle': cycle, 'StableRequiredFieldsInTwoPasses': stable,
                  'CountsAgreeWithClosedPair': all(value['AllocatedTypedSlots'] == observation['AllocatedDependentSlots'] for value in passes),
                  'QualifiedNativeRead': valid, 'EvidenceFiles': files,
                  'ComparisonScope': 'Complete depstore headers/slots, descriptors, live-thread root chains, TLS bodies and exact pool CWT entries; unrelated batched-read gap bytes excluded.'}
        (folder / 'Capture.json').write_text(json.dumps(record, indent=2), encoding='utf-8')
        assert valid, ('Unstable or inconsistent closed native ownership capture', record)
        return second, record


def _root_identity(root):
    return (root['RootKind'], root.get('ThreadAddress'), root.get('ThreadId'), root.get('RegistryIndex'))


def _row(root, module):
    return root.get('TypeStorageRows', {}).get(module, root.get('TypeStorageRows', {}).get(str(module)))


def _pool_field(root, descriptor):
    module, index = descriptor['ModuleIndex'], descriptor['TypeIndex']
    if module is None or module < 0 or index < 0:
        return {'TypeBase': 0, 'Target': 0}
    row = _row(root, module)
    if row is None or index >= row['Length'] or row['DataWords'][index] == 0:
        return {'TypeBase': 0, 'Target': 0}
    fields = [field for field in root['PoolFields'] if field['ClosedT'] == descriptor['ClosedT']]
    assert len(fields) == 1 and fields[0]['TypeBase'] == row['DataWords'][index]
    return fields[0]


class ClosedCycleOwnershipTracker:
    """Fixed non-TLS baseline with exact first-use runtime TLS ownership checks."""
    def __init__(self):
        self.previous = None
        self.baseline_non_tls = None
        self.baseline_tombstones = None
        self.known_tls = {}

    def qualify(self, value):
        runtime = value['RuntimeTlsPools']
        descriptors = {entry['ClosedT']: entry for entry in runtime['Descriptors']}
        assert descriptors and len(descriptors) == runtime['DescriptorCount']
        roots = {_root_identity(root): root for root in runtime['ThreadRootSnapshots']}
        assert len(roots) == len(runtime['ThreadRootSnapshots']), 'Duplicate live-thread identity'
        slots = {slot['SlotAddress']: slot for slot in value['Slots']}
        assert len(slots) == value['AllocatedTypedSlots'] == value['AliveCounterSum']
        current = {}
        for body in runtime['TlsArrayBodies']:
            slot = slots[body['SlotAddress']]
            assert slot['TargetObject'].get('EEType') == value['SupportedTlsArrayEEType'], 'Fake TLS object type'
            assert slot['Target'] == body['TlsArray'] and slot['Target'] and slot['Dependent'] == 0
            assert not slot['ClearedTombstone'] and body['Body']['Length'] == 27
            assert len(body['ThreadOwners']) == len(body['PoolRegistryOwners']) == 1, 'TLS owner missing or aliased'
            owner, pool = body['ThreadOwners'][0], body['PoolRegistryOwners'][0]
            assert owner['ClosedT'] in descriptors and owner['ClosedT'] == pool['ClosedT'], 'Wrong closed-T pool ownership'
            identity = _root_identity(owner)
            assert identity in roots, 'TLS owner is not a fixed live thread/bootstrap root'
            descriptor = descriptors[owner['ClosedT']]
            field = _pool_field(roots[identity], descriptor)
            assert field['Target'] == owner['Target'] == slot['Target']
            assert field['TypeBase'] == owner['TypeBase']
            assert descriptor['ModuleIndex'] == owner['ModuleIndex'] and descriptor['TypeIndex'] == owner['TypeIndex']
            assert pool['DependentHandle'] == slot['SlotAddress'] and pool['SlotTarget'] == slot['Target'] and pool['SlotDependent'] == 0
            assert pool['HashCode'] >= 0, 'Free CWT entry cannot establish runtime ownership'
            key = (owner['ClosedT'], *identity)
            assert key not in current, 'More than one TLS array for the same live T/thread'
            current[key] = {'SlotAddress': slot['SlotAddress'], 'Target': slot['Target'],
                            'TypeBase': owner['TypeBase'], 'ClosedT': owner['ClosedT'], 'RootIdentity': identity}
        tls_addresses = {entry['SlotAddress'] for entry in current.values()}
        assert len(tls_addresses) == len(current), 'One runtime TLS slot is aliased by multiple owners'
        assert len({entry['Target'] for entry in current.values()}) == len(current), 'A runtime TLS target is aliased by multiple owners'
        non_tls = {address: (slot['Target'], slot['Dependent']) for address, slot in slots.items() if address not in tls_addresses}
        tombstones = {address for address, slot in slots.items() if slot['ClearedTombstone']}
        initialized = []
        if self.previous is None:
            self.baseline_non_tls = non_tls
            self.baseline_tombstones = tombstones
        else:
            old_runtime = self.previous['RuntimeTlsPools']
            old_descriptors = {entry['ClosedT']: entry for entry in old_runtime['Descriptors']}
            assert descriptors == old_descriptors, 'Compiled native pool descriptors changed'
            old_roots = {_root_identity(root): root for root in old_runtime['ThreadRootSnapshots']}
            assert set(roots) == set(old_roots), 'A fixed live thread exited, appeared or changed identity'
            assert non_tls == self.baseline_non_tls, 'Unexplained non-TLS native slot, deletion or pair change'
            assert tombstones == self.baseline_tombstones, 'A native tombstone appeared or changed'
            for key, previous in self.known_tls.items():
                assert current.get(key) == previous, 'Existing TLS root/slot was lost or replaced'
            for identity, root in roots.items():
                old = old_roots[identity]
                old_modules, modules = old.get('ModuleStorage'), root.get('ModuleStorage')
                if old_modules:
                    assert modules and modules['Length'] >= old_modules['Length'], 'TLS module storage shrank or disappeared'
                    for module, old_address in enumerate(old_modules['DataWords']):
                        new_address = modules['DataWords'][module]
                        if old_address != new_address and old_address:
                            old_row, new_row = _row(old, module), _row(root, module)
                            assert old_row and new_row and new_row['Length'] >= old_row['Length']
                            assert new_row['DataWords'][:old_row['Length']] == old_row['DataWords'], 'TLS row resize did not preserve every prior field'
                for generic, descriptor in descriptors.items():
                    before, after = _pool_field(old, descriptor), _pool_field(root, descriptor)
                    if before['TypeBase']:
                        assert after['TypeBase'] == before['TypeBase'], 'Existing compiled TLS type-base was replaced'
                    if before['Target']:
                        assert after['Target'] == before['Target'], 'Existing compiled TLS field was replaced or cleared'
            for key, entry in current.items():
                if key in self.known_tls:
                    continue
                identity = tuple(entry['RootIdentity'])
                before = _pool_field(old_roots[identity], descriptors[entry['ClosedT']])
                assert before['Target'] == 0, 'Additional TLS slot aliases an already initialized runtime root'
                initialized.append({'ClosedT': entry['ClosedT'], 'RootIdentity': identity,
                    'PreviousTypeBase': before['TypeBase'], 'PreviousTarget': 0, 'NewTypeBase': entry['TypeBase'],
                    'NewTarget': entry['Target'], 'ExactNativeSlot': entry['SlotAddress']})
        self.known_tls.update(current)
        self.previous = value
        return {'Passed': True, 'AllocatedGlobalDependentSlots': value['AllocatedTypedSlots'],
                'ExcludedProvenLiveRuntimeTlsSlots': len(current), 'FixedNonTlsSlots': len(non_tls),
                'FixedTombstones': len(tombstones), 'ProvenFirstUseRuntimeTlsInitializations': initialized,
                'AllowedUnknownGrowth': 0, 'WarmBaselineRolled': False}