Add verified packed sector-table AOT overlay method

docs/AOT_SHARDING.md

@@ -59,9 +59,11 @@ Profiles declare these reusable methods:
 | `disc_hashes` | Hash the cue's original data track before extracting. |
 | `images[].method = fixed_address_files` | Whole original files placed at a verified `load_addr`. Optional duplicate-leaf checks require identical bytes. |
 | `images[].method = psx_exe` | Read the load address and image from a PS-X EXE header; the generic extractor may split resident and overlay floors. |
+| `images[].method = packed_sector_members` | Decode u32 count/offset sector descriptors across ordered payload files. Verify every member classification and requested full-payload coverage; use loader-established addresses for executable members. |
 | `checks[].method = words` | Verify loader instructions or descriptors at explicit file offsets / virtual addresses. |
 | `checks[].method = pointer_strings` | Verify a pointer-indexed filename table against exact expected strings. |
 | `checks[].method = bcd_extent_table` | Verify an indexed BCD-MSF/size table against an ISO file's actual extent. |
+| `checks[].method = adjacent_files` | Require the named ISO extents to be sector aligned and physically adjacent, in order. |
 | `compositions` | Combine declared simultaneous producers in address order. `max_gap` bounds their separation; gap and alignment bytes remain unowned. |
 | `expected_records` | Stop on inventory drift, including unexpected additions requiring review. Every configured image must be represented. |
 | `strict_bounds` | Require emission to remain inside established producer intervals. |
@@ -73,7 +75,9 @@ into a proven function. There is no recipe option to ignore a failed compiler.
 
 Each recipe compiles in an isolated cache. `tools/audit_aot_cache.py` checks every
 native pair's ABI/exports, every manifest guard against known original bytes,
-and nonzero native coverage for every recipe. Staging copies precisely the
+and nonzero native coverage for every recipe. Loader entry points supplied by
+the profile must also have native entries guarded by that image's original bytes.
+Staging copies precisely the
 audited platform pairs, verifies hashes again, rejects extras, and emits
 `AOT_CACHE_AUDIT.json` without game bytes. A missing disc, changed loader word,
 missing image, compiler failure, empty recipe coverage or failed audit stops
@@ -89,6 +93,20 @@ Current consumers:
   MAIN.EXE and 28 raw images, including all 22 area overlays. Their loader tables
   establish different regional addresses; composition uses the same shared
   method (including the Italian gap). US START.BIN is explicitly accounted for.
+- Ape Escape USA: `ApeEscapeRecomp/aot/overlays.json`, 47 established packed
+  sector-table overlays plus two minigame PS-X EXEs. Two additional archive
+  members have unresolved load paths and are explicitly excluded.
+
+For `packed_sector_members`, declare `table_file`, ordered `payload_files`,
+`sector_size` (default 2048), `offset_bits` (default 20), and optional
+`table_offset`. A descriptor's low bits are the logical sector offset and its
+remaining high bits are the sector count. Members cannot cross payload-file
+boundaries. `inventory_range` gives inclusive `first`/`last` descriptor indices;
+every index must occur once in either `members` or `excluded_members`.
+Executable members carry `index`, `load_addr`, and optionally verified `entries`.
+Exclusions require a `reason`. `cover_payloads` names files that this entire
+classified range must cover exactly, without gaps or overlaps. The shared reader
+is `tools/packed_sector_table.py`; heuristic discovery uses that same parser.
 
 See `tools/tests/test_aot_overlay_pipeline.py` for synthetic method and release
 failure tests. A new title should add a profile when these contracts fit; add

tools/aot_overlay_pipeline.py

@@ -19,6 +19,7 @@ import zlib
 
 import compile_overlays as compiler
 from aot_overlay_spike import extract_generic as extractor
+from packed_sector_table import extract_members as extract_sector_members
 
 FRAMEWORK = Path(__file__).resolve().parents[1]
 
@@ -61,6 +62,12 @@ class Disc:
 def verify_evidence(disc, checks):
     """Reusable binary-word, pointer-string, and BCD extent table identifiers."""
     for check in checks:
+        if check['method'] == 'adjacent_files':
+            files = [disc.files[name.upper()] for name in check['files']]
+            require(all(size % 2048 == 0 and lba + size // 2048 == next_lba
+                        for (lba, size), (next_lba, _) in zip(files, files[1:])),
+                    'Original disc files are not sector-adjacent')
+            continue
         data = disc.read(check['file'])
         origin = number(check.get('file_offset', 0))
         base = number(check.get('base', 0))
@@ -93,10 +100,54 @@ def verify_evidence(disc, checks):
             raise ValueError(f'Unknown evidence method: {method}')
 
 
+def sector_sources(disc, spec):
+    """Verified member inventory, with explicit exclusions and extent checks."""
+    first, last = (number(spec['inventory_range'][key]) for key in ('first', 'last'))
+    require(0 <= first <= last, 'Invalid sector inventory range')
+    configured = {number(item['index']): item for item in spec['members']}
+    excluded = {number(item['index']): item for item in spec.get('excluded_members', [])}
+    require(len(configured) == len(spec['members']) and
+            len(excluded) == len(spec.get('excluded_members', [])), 'Duplicate member inventory')
+    require(not configured.keys() & excluded.keys() and
+            configured.keys() | excluded.keys() == set(range(first, last + 1)),
+            'Sector inventory has missing or conflicting classifications')
+    require(all(item.get('reason', '').strip() for item in excluded.values()),
+            'Excluded sector member needs a reason')
+    members = extract_sector_members(disc.read(spec['table_file']),
+                    [(name.upper(), disc.read(name)) for name in spec['payload_files']],
+                    range(first, last + 1), sector_size=number(spec.get('sector_size', 2048)),
+                    offset_bits=number(spec.get('offset_bits', 20)),
+                    table_offset=number(spec.get('table_offset', 0)))
+    for name in spec.get('cover_payloads', []):
+        spans = sorted((member['source_offset'], len(member['body'])) for member in members
+                       if member['source_file'] == name.upper())
+        cursor = 0
+        for offset, size in spans:
+            require(offset == cursor, 'Sector inventory has a gap or overlapping members')
+            cursor += size
+        require(cursor == len(disc.read(name)), 'Sector inventory does not cover required payload')
+    sources = []
+    for member in members:
+        item = configured.get(member['index'])
+        if item is None:
+            continue
+        base = number(item['load_addr'])
+        require(0x80000000 <= base < base + len(member['body']) <= 0x80200000,
+                'Sector image outside RAM')
+        name = f"{spec['table_file'].upper()}:ENTRY_{member['index']:04X}"
+        sources.append(dict(name=name, base=base, body=member['body'],
+                            spec={**spec, **item}, source_offset=member['source_offset'],
+                            source_file=member['source_file'], aliases=[]))
+    return sources
+
+
 def positioned_sources(disc, specifications):
     sources = []
     for spec in specifications:
         method = spec['method']
+        if method == 'packed_sector_members':
+            sources.extend(sector_sources(disc, spec))
+            continue
         for name in spec['files']:
             body = disc.read(name)
             offset = 0
@@ -140,13 +191,8 @@ def match_sources(record, sources):
     return matches
 
 
-def make_fixed_record(source, disc):
+def declared_entries(source, disc):
     body, base, spec = source['body'], source['base'], source['spec']
-    require(spec.get('allow_missing'), f"Generic extractor missed required image: {source['name']}")
-    direct = set(extractor.direct_jal_roots(body, base))
-    seeds = set(extractor.prologues(body, base)) | extractor.frameless_leaf_entries(body, base)
-    if spec.get('supplemental_entries', True):
-        seeds |= extractor.supplemental_callable_seeds(body, base)
     entries = set()
     if 'entry_word' in spec:
         item = spec['entry_word']
@@ -155,6 +201,17 @@ def make_fixed_record(source, disc):
     entries.update(number(x) for x in spec.get('entries', []))
     require(all(base <= entry < base + len(body) and entry % 4 == 0 for entry in entries),
             f"Entry outside image: {source['name']}")
+    return entries
+
+
+def make_fixed_record(source, disc):
+    body, base, spec = source['body'], source['base'], source['spec']
+    require(spec.get('allow_missing'), f"Generic extractor missed required image: {source['name']}")
+    direct = set(extractor.direct_jal_roots(body, base))
+    seeds = set(extractor.prologues(body, base)) | extractor.frameless_leaf_entries(body, base)
+    if spec.get('supplemental_entries', True):
+        seeds |= extractor.supplemental_callable_seeds(body, base)
+    entries = declared_entries(source, disc)
     seeds |= entries
     require(seeds, f"No static entries: {source['name']}")
     page, data = extractor.page_aligned_region(base, body)
@@ -194,10 +251,21 @@ def prepare(profile, disc, records, output):
             if (lo, hi) == (source['base'], source['base'] + len(source['body'])):
                 singles[source['name']] = record
     for source in sources:
-        if source['spec']['method'] == 'fixed_address_files' and source['name'] not in singles:
+        if source['spec']['method'] in ('fixed_address_files', 'packed_sector_members') and source['name'] not in singles:
             record = make_fixed_record(source, disc)
             singles[source['name']] = record
             records.append(record)
+    # Loader-established exports remain required even when heuristic discovery
+    # found the image without finding that particular callable entry.
+    for source in sources:
+        entries = declared_entries(source, disc)
+        if entries:
+            require(source['name'] in singles, 'Declared entries need a complete source recipe')
+            record = singles[source['name']]
+            for key in ('function_entry_pcs', 'dispatch_entry_pcs', 'static_dispatch_entry_pcs', 'seeds'):
+                record[key] = [hex(pc) for pc in sorted(entries | {number(pc) for pc in record[key]})]
+            record['static_discovery_entry_pcs'] = [hex(pc) for pc in sorted(entries |
+                {number(pc) for pc in record.get('static_discovery_entry_pcs', [])})]
     by_name = {source['name']: source for source in sources}
     keys = {(number(r['load_addr']), r['bytes_b64']) for r in records}
     for composition in profile.get('compositions', []):
@@ -231,6 +299,8 @@ def prepare(profile, disc, records, output):
         path = output / 'runtime-inputs' / f'{index:03d}.json'
         write_json(path, [record])
         jobs.append(dict(name=f'{index:03d}-' + '+'.join(names), sources=names,
+                         required_entries=sorted({entry for source, _, _ in matches
+                                                  for entry in declared_entries(source, disc)}),
                          known_ranges=bounds, input=str(path.resolve()), load_addr=hex(load),
                          size=len(data), sha256=hashlib.sha256(data).hexdigest(),
                          recipe_sha256=hashlib.sha256(json.dumps(record, sort_keys=True).encode()).hexdigest(),

tools/aot_overlay_spike/extract_generic.py

@@ -30,6 +30,7 @@ import argparse, os, sys, json, base64, struct, subprocess, tempfile, re, binasc
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
 import extract_overlays as eo
 import compile_overlays as co
+from packed_sector_table import extract_members as extract_sector_members
 try:
     import tomllib
 except ImportError:
@@ -565,25 +566,18 @@ def hed_companion_members(disc, files):
         if not runs:
             continue
 
-        logical=[]; cursor=0
-        for path,lba,size in companions:
-            body=disc.read_file_bytes(lba,size)
-            logical.append((cursor,cursor+size//0x800,path,body))
-            cursor+=size//0x800
+        payloads=[(path,disc.read_file_bytes(lba,size)) for path,lba,size in companions]
         members=[]; seen=set()
         for table_index,sector,count in runs:
             key=(sector,count)
             if key in seen:
                 continue
             seen.add(key)
-            owner=next((x for x in logical
-                        if x[0] <= sector and sector+count <= x[1]),None)
-            if owner is None:
+            try:
+                member=extract_sector_members(hed,payloads,[table_index])[0]
+            except ValueError:
                 continue
-            lo,hi,path,body=owner
-            off=(sector-lo)*0x800
-            member=body[off:off+count*0x800]
-            members.append((table_index,sector*0x800,member))
+            members.append((table_index,member['logical_offset'],member['body']))
         if members:
             groups.append({
                 'hed_path':hed_path,

tools/audit_aot_cache.py

@@ -93,6 +93,8 @@ def main():
     if args.expected_pairs is not None:
         assert len(pairs) == args.expected_pairs, (len(pairs), args.expected_pairs)
     image_coverage = []
+    required = {job['name']: {entry & 0x1fffffff for entry in job.get('required_entries', [])}
+                for job in jobs}
     for name, load, data, bounds in recipes:
         compatible = set()
         for physical, ids in native_ids:
@@ -105,7 +107,10 @@ def main():
                               for entry, crc, ranges in compiler.current_variant_func_ids(
                                   bounded, data, load, len(data)))
         assert compatible, f'No guarded native entries for input image: {name}'
+        served = {item[0] & 0x1fffffff for item in compatible}
+        assert required[name] <= served, f'Missing loader-established entries for {name}: {sorted(required[name] - served)}'
         image_coverage.append(dict(name=name,
+                                   required_entries_verified=len(required[name]),
                                    native_entries=len({item[0] for item in compatible}),
                                    guarded_variants=len(compatible)))
     receipt = dict(game_id=game_id, cache_tag=tag, flavor=args.flavor,

tools/packed_sector_table.py

@@ -0,0 +1,44 @@
+"""Strict extraction of u32 packed (sector count, sector offset) descriptors.
+
+Payload files form one logical byte stream, in the caller's declared order.
+This is a container method; identifying executable members and their load
+addresses belongs to independently verified loader evidence.
+"""
+import struct
+
+
+def extract_members(table, payloads, indices, *, sector_size=2048, offset_bits=20,
+                    table_offset=0):
+    if sector_size <= 0 or sector_size & (sector_size - 1):
+        raise ValueError('Sector size must be a positive power of two')
+    if not 1 <= offset_bits < 32 or table_offset < 0 or table_offset % 4:
+        raise ValueError('Invalid packed sector descriptor layout')
+    if len({name for name, _ in payloads}) != len(payloads):
+        raise ValueError('Duplicate sector payload')
+    spans, cursor = [], 0
+    for name, body in payloads:
+        if not body or len(body) % sector_size:
+            raise ValueError('Sector payload must be nonempty and sector aligned')
+        spans.append((cursor, cursor + len(body), name, body))
+        cursor += len(body)
+    members, seen = [], set()
+    for index in indices:
+        if index in seen or index < 0:
+            raise ValueError('Duplicate or negative sector table index')
+        seen.add(index)
+        at = table_offset + index * 4
+        if at > len(table) - 4:
+            raise ValueError('Sector descriptor outside table')
+        descriptor = struct.unpack_from('<I', table, at)[0]
+        start = (descriptor & ((1 << offset_bits) - 1)) * sector_size
+        size = (descriptor >> offset_bits) * sector_size
+        if not size:
+            raise ValueError('Empty sector member')
+        owner = next((span for span in spans if span[0] <= start < start + size <= span[1]), None)
+        if owner is None:
+            raise ValueError('Sector member crosses or exceeds payload bounds')
+        lo, _, name, body = owner
+        offset = start - lo
+        members.append(dict(index=index, logical_offset=start, source_file=name,
+                            source_offset=offset, body=body[offset:offset + size]))
+    return members

tools/tests/test_aot_overlay_pipeline.py

@@ -22,6 +22,42 @@ class FakeDisc:
 
 
 class AotMethodsTest(unittest.TestCase):
+    def test_sector_inventory_covers_payload_and_accounts_for_exclusions(self):
+        disc = FakeDisc({'INDEX': struct.pack('<III', 0x100000, 0x100001, 0x100002),
+                         'DATA': b'A' * 2048, 'CODE': b'B' * 2048 + b'C' * 2048})
+        spec = dict(method='packed_sector_members', table_file='INDEX',
+                    payload_files=['DATA', 'CODE'], inventory_range=dict(first=1, last=2),
+                    cover_payloads=['CODE'], members=[dict(index=1, load_addr='0x80100000')],
+                    excluded_members=[dict(index=2, reason='Load destination unresolved')])
+        sources = pipeline.positioned_sources(disc, [spec])
+        self.assertEqual([(s['name'], s['body'], s['source_offset']) for s in sources],
+                         [('INDEX:ENTRY_0001', b'B' * 2048, 0)])
+        with self.assertRaisesRegex(ValueError, 'classifications'):
+            pipeline.positioned_sources(disc, [{**spec, 'excluded_members': []}])
+        disc.data['INDEX'] = struct.pack('<III', 0x100000, 0x100001, 0x100001)
+        with self.assertRaisesRegex(ValueError, 'overlapping'):
+            pipeline.positioned_sources(disc, [spec])
+
+    def test_sector_bounds_and_descriptor_layout_fail_closed(self):
+        for descriptor, indices, options, error in [
+                (0x200000, [0], {}, 'crosses'), (0x100002, [0], {}, 'bounds'),
+                (0, [0], {}, 'Empty'), (0x100000, [1], {}, 'outside table'),
+                (0x100000, [0, 0], {}, 'Duplicate'),
+                (0x100000, [0], dict(offset_bits=32), 'layout'),
+                (0x100000, [0], dict(sector_size=3), 'power of two')]:
+            with self.subTest(error=error), self.assertRaisesRegex(ValueError, error):
+                pipeline.extract_sector_members(struct.pack('<I', descriptor),
+                    [('A', bytes(2048)), ('B', bytes(2048))], indices, **options)
+
+    def test_adjacent_disc_extents_are_verified(self):
+        disc = FakeDisc({'INDEX': bytes(2048), 'BODY': bytes(4096)})
+        disc.files = {'INDEX': (10, 2048), 'BODY': (11, 4096)}
+        check = dict(method='adjacent_files', files=['INDEX', 'BODY'])
+        pipeline.verify_evidence(disc, [check])
+        disc.files['BODY'] = (12, 4096)
+        with self.assertRaisesRegex(ValueError, 'sector-adjacent'):
+            pipeline.verify_evidence(disc, [check])
+
     def test_empty_primary_needs_every_current_byte_root_and_preserves_other_errors(self):
         compiler = pipeline.compiler
         pending = [('image', 0x100000, 0x80100000, 16, b'bytes', {0x80100000, 0x80100008})]