tricu

unit.py (100932B)

---

 #!/usr/bin/env python3
 """Fast host-side policy, provenance, RSS and native-scope controls; no builds."""
 
 # native_header_equivalence_check
 import hashlib
 
 import json
 
 from pathlib import Path
 
 import tempfile
 
 import unittest
 
 from native_header_equivalence import native_header_equivalence, digest
 
 class HeaderEquivalenceTests(unittest.TestCase):
     root_kind = 'unit'
 
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory()
         self.addCleanup(self.temporary.cleanup)
         root = Path(self.temporary.name)
         self.repo, self.generation = root / 'repo', root / 'generation'
         self.repo.mkdir(); self.generation.mkdir()
         (self.generation / 'overlay').mkdir()
         (self.generation / 'cas/cache/source-build').mkdir(parents=True)
         self.map = self.repo / 'map.v1'
         old_map = ('arboricx.source-map.v1\nunit\tbase\tbase\tbase.tri\t-\n' +
                    self.root_kind + '\tsource_map\tsource.map\troot.tri\tbase\n')
         self.map.write_text(old_map.replace('unit\t', 'native\t').replace('reexport\t', 'native\t'))
         (self.generation / 'source-map.v1').write_text(old_map)
         raw = {'base': b'x = t\n', 'source_map': b'!import "base" !Local\nf x = x\n'}
         staged = {'base': b'!export *\nx = t\n\n',
                   'source_map': (b'!import base *\n!export *\n' +
                                  (b'!export * from base\n' if self.root_kind == 'reexport' else b'') +
                                  b'f x = x\n\n')}
         compiler = '1' * 64
         provenance = dict(map_sha256=hashlib.sha256(old_map.encode()).hexdigest(),
                           sources={k: hashlib.sha256(v).hexdigest() for k, v in raw.items()},
                           compiler_bundle_sha256=compiler)
         (self.generation / 'inputs.json').write_text(json.dumps(provenance))
         (self.generation / 'status.json').write_text(json.dumps(dict(complete=True, root='source_map')))
         for alias, path in [('base', 'base.tri'), ('source_map', 'root.tri')]:
             self.repo.joinpath(path).write_bytes(staged[alias])
             self.generation.joinpath('overlay', alias + '.tri').write_bytes(staged[alias])
             tree = self.put('arboricx.tree-term.v1', b'\0')
             manifest = self.put('arboricx.module-manifest.v1', ('arboricx.module-manifest.v1\nexport\tmain\tarboricx.tree-term.v1\t' + tree + '\tarboricx.abi.tree.v1\n').encode())
             rows = ['source\t' + digest('arboricx.source.v1', staged[alias]),
                     'compiler\t' + compiler,
                     'options\t' + digest('arboricx.build-options.v1', b'source-module.reference.v1')]
             receipt = self.put('arboricx.build-receipt.v1', ('arboricx.build-receipt.v1\n' + '\n'.join(rows) + '\noutput\t' + manifest + '\n').encode())
             key = digest('arboricx.source-build-key.v1', ('arboricx.source-build-key.v1\n' + '\n'.join(rows) + '\n').encode())
             self.generation.joinpath('cas/cache/source-build', alias).write_text('\t'.join(['arboricx.source-build-cache.v1', key, manifest, receipt]))
 
     def put(self, domain, payload):
         pin = digest(domain, payload)
         path = self.generation / 'cas/objects' / pin[:3] / pin
         path.parent.mkdir(parents=True, exist_ok=True); path.write_bytes(payload)
         return pin
 
     def check(self):
         return native_header_equivalence(self.generation, self.repo, self.map)
 
     def test_valid_staged_bytes(self):
         rows, proof = self.check()
         self.assertEqual(proof['changed'], ['base', 'source_map'])
         self.assertEqual(len(rows), 2)
 
     def test_changed_body(self):
         with (self.repo / 'root.tri').open('a') as f: f.write('-- body change\n')
         with self.assertRaisesRegex(ValueError, 'staged bytes'): self.check()
 
     def test_changed_namespace(self):
         p = self.repo / 'root.tri'; p.write_text(p.read_text().replace('!import base *', '!import base as B'))
         with self.assertRaisesRegex(ValueError, 'staged bytes'): self.check()
 
     def test_changed_exports(self):
         p = self.repo / 'root.tri'; p.write_text(p.read_text().replace('!export *', '!export f'))
         with self.assertRaisesRegex(ValueError, 'staged bytes|invalid native export'): self.check()
 
     def test_changed_map_path(self):
         (self.repo / 'other.tri').write_bytes((self.repo / 'base.tri').read_bytes())
         self.map.write_text(self.map.read_text().replace('base.tri', 'other.tri'))
         with self.assertRaisesRegex(ValueError, 'edges, paths'): self.check()
 
     def test_incomplete_generation(self):
         (self.generation / 'status.json').write_text('{"complete":false,"root":"source_map"}')
         with self.assertRaisesRegex(ValueError, 'completed'): self.check()
 
     def test_overlay_receipt_corruption(self):
         # Keep current source equal to the modified overlay: the receipt still
         # authenticates the old compiler input and must reject this pair.
         for p in [self.repo / 'root.tri', self.generation / 'overlay/source_map.tri']:
             with p.open('a') as f: f.write('-- forged overlay\n')
         with self.assertRaisesRegex(ValueError, 'receipt mismatch'): self.check()
 
     def test_checkpoint_key_corruption(self):
         p = self.generation / 'cas/cache/source-build/source_map'
         parts = p.read_text().split('\t'); parts[1] = '2' * 64; p.write_text('\t'.join(parts))
         with self.assertRaisesRegex(ValueError, 'key mismatch'): self.check()
 
 class ReexportHeaderEquivalenceTests(HeaderEquivalenceTests):
     # Exercise the same fail-closed controls with the prelude's re-export policy.
     root_kind = 'reexport'
 
 
 # native_library_migration_check
 import unittest
 
 from native_library_migration import native_candidate, alias, terminal_export, reexport_candidate
 
 class HeaderTests(unittest.TestCase):
     def test_local_and_namespace_preserved(self):
         body = b'-- body\nf x = N.id x\n'
         raw = b'!import "prelude" !Local\n!import "source.core" N\n' + body
         candidate, imports = native_candidate(raw, {'prelude': 'prelude', 'source.core': 'source_core'})
         self.assertEqual(candidate, b'!import prelude *\n!import source_core as N\n!export *\n'+body)
         self.assertEqual(imports, [('prelude', '!Local'), ('source_core', 'N')])
 
     def test_comments_before_import_preserved(self):
         raw = b'-- header comment\n!import "prelude" !Local\nx = t\n'
         candidate, _ = native_candidate(raw, {'prelude': 'prelude'})
         self.assertTrue(candidate.endswith(b'-- header comment\nx = t\n'))
 
     def test_explicit_transitive_dependency_is_recorded(self):
         raw = b'!import "patterns" !Local\nmain = equal? t t\n'
         candidate, imports = native_candidate(raw, {'patterns': 'patterns'}, [('prelude', '!Local')])
         self.assertEqual(imports, [('prelude', '!Local'), ('patterns', '!Local')])
         self.assertTrue(candidate.startswith(b'!import prelude *\n!import patterns *\n'))
         self.assertTrue(candidate.endswith(b'main = equal? t t\n'))
 
     def test_unknown_dependency_rejected(self):
         with self.assertRaisesRegex(ValueError, 'unknown dependency'):
             native_candidate(b'!import "missing" !Local\nx = t\n', {})
 
     def test_malformed_import_rejected(self):
         with self.assertRaisesRegex(ValueError, 'unsupported legacy'):
             native_candidate(b'!import "prelude"\nx = t\n', {'prelude': 'prelude'})
 
     def test_native_policy_not_widened(self):
         raw = b'!import dep as D\n!export f\nf = D.f\n'
         candidate, imports = native_candidate(raw, {'dep': 'dep'})
         self.assertEqual(candidate, raw)
         self.assertEqual(imports, [('dep', 'D')])
 
     def test_unknown_native_dependency_rejected(self):
         with self.assertRaisesRegex(ValueError, 'unknown native'):
             native_candidate(b'!import dep *\n!export f\nf = t\n', {'other': 'other'})
 
     def test_terminal_selector_requires_explicit_exact_symbol(self):
         raw = b'program x = x\nprogram\n\n'
         self.assertEqual(terminal_export(raw, 'program'), b'program x = x\n\n')
         with self.assertRaisesRegex(ValueError, 'exact bare'):
             terminal_export(b'program x = x\nprogram t\n', 'program')
 
     def test_terminal_selector_requires_authored_definition(self):
         with self.assertRaisesRegex(ValueError, 'authored definition'):
             terminal_export(b'program\n', 'program')
 
     def test_terminal_selector_rejects_expression_names(self):
         with self.assertRaisesRegex(ValueError, 'invalid terminal'):
             terminal_export(b'program t\n', 'program t')
 
     def test_reexport_directives_precede_body_comments(self):
         raw = b'-- example\n!import "dep" !Local\n'
         candidate, imports = native_candidate(raw, {'dep':'dep'})
         result = reexport_candidate(raw, candidate, imports)
         self.assertIn(b'!export *\n!export * from dep\n-- example', result)
         self.assertEqual(reexport_candidate(result,result,imports),result)
 
     def test_reexport_policy_rejects_authored_definitions(self):
         with self.assertRaisesRegex(ValueError,'only comments'):
             reexport_candidate(b'f = t\n',b'!export *\nf = t\n',[])
 
     def test_reexport_policy_rejects_namespace_widening(self):
         with self.assertRaisesRegex(ValueError,'local imports'):
             reexport_candidate(b'-- comment\n',b'!export *\n',[('dep','N')])
 
     def test_flat_aliases(self):
         self.assertEqual(alias('source.export-cache'), 'source_export_cache')
 
 
 # process_rss_check
 import subprocess
 
 import unittest
 
 from types import SimpleNamespace
 
 from unittest.mock import patch
 
 from process_rss import ChildRSS, status_fields
 
 class ChildRSSTests(unittest.TestCase):
     def monitor(self, candidates=(7, 1001)):
         monitor = ChildRSS.__new__(ChildRSS)
         def still_running(timeout):
             raise subprocess.TimeoutExpired('fake-child', timeout)
         monitor.proc = SimpleNamespace(pid=7, poll=lambda: None, wait=still_running)
         monitor.parent, monitor.namespace_index, monitor.pid = 1000, 1, None
         monitor.candidates = lambda: iter(candidates)
         return monitor
 
     def test_namespace_translation_ignores_wrong_host_pid(self):
         def read(path):
             if str(path) == '/proc/7/status':
                 return 'Pid: 7\nPPid: 2\nNSpid: 7\nVmRSS: 999999 kB\n'
             return 'Pid: 1001\nPPid: 1000\nNSpid: 1001 7\nVmRSS: 512 kB\n'
         monitor = self.monitor()
         with patch('pathlib.Path.read_text', read):
             self.assertEqual(monitor.sample(), 512)
             self.assertEqual(monitor.pid, 1001)
 
     def test_missing_rss_fails_closed(self):
         with patch('pathlib.Path.read_text', return_value='Pid: 1001\nPPid: 1000\nNSpid: 1001 7\n'):
             with self.assertRaisesRegex(RuntimeError, 'cannot monitor'):
                 self.monitor((1001,)).sample()
 
     def test_inaccessible_rss_fails_closed(self):
         with patch('pathlib.Path.read_text', side_effect=FileNotFoundError):
             with self.assertRaisesRegex(RuntimeError, 'cannot monitor'):
                 self.monitor().sample()
 
     def test_cached_pid_reuse_fails_closed(self):
         monitor = self.monitor(); monitor.pid = 1001
         with patch('pathlib.Path.read_text', return_value='Pid: 1001\nPPid: 1000\nNSpid: 1001 8\nVmRSS: 10 kB\n'):
             with self.assertRaisesRegex(RuntimeError, 'cannot monitor'):
                 monitor.sample()
 
     def test_exiting_leader_uses_shared_thread_memory(self):
         from pathlib import Path
         monitor = self.monitor((1001,))
         def read(path):
             if str(path) == '/proc/1001/status':
                 return 'Pid: 1001\nPPid: 1000\nNSpid: 1001 7\n'
             return 'Tgid: 1001\nVmRSS: 321 kB\n'
         with patch('pathlib.Path.read_text', read), patch('pathlib.Path.glob', return_value=[Path('/proc/1001/task/1002/status')]):
             self.assertEqual(monitor.sample(), 321)
 
     def test_identified_running_exit_with_no_address_space(self):
         monitor = self.monitor((1001,))
         def read(path):
             if str(path).endswith('/statm'):
                 return '0 0 0 0 0 0 0\n'
             return 'Pid: 1001\nPPid: 1000\nNSpid: 1001 7\nState: R (running)\n'
         with patch('pathlib.Path.read_text', read), patch('pathlib.Path.glob', return_value=[]):
             self.assertEqual(monitor.sample(), 0)
 
     def test_identified_zombie_can_finish_kernel_cleanup(self):
         monitor = self.monitor((1001,))
         with patch('pathlib.Path.read_text', return_value='Pid: 1001\nPPid: 1000\nNSpid: 1001 7\nState: Z (zombie)\n'), patch('pathlib.Path.glob', return_value=[]):
             self.assertIsNone(monitor.sample())
 
     def test_exited_child_has_no_sample(self):
         monitor = self.monitor(); monitor.proc.poll = lambda: 0
         self.assertIsNone(monitor.sample())
 
     def test_address_space_disappears_just_before_reap(self):
         monitor = self.monitor()
         monitor.proc.wait = lambda timeout: 0
         with patch('pathlib.Path.read_text', side_effect=FileNotFoundError):
             self.assertIsNone(monitor.sample())
 
     def test_real_child_is_visible(self):
         proc = subprocess.Popen(['sleep', '2'])
         try:
             monitor = ChildRSS(proc)
             sample = monitor.sample()
             self.assertIsNotNone(sample)
             self.assertGreater(sample, 0)
             self.assertTrue(monitor.matches(status_fields(__import__('pathlib').Path(f'/proc/{monitor.pid}/status').read_text())))
         finally:
             proc.terminate(); proc.wait()
 
 
 # research_session_check
 import hashlib
 
 from pathlib import Path
 
 import subprocess
 
 import tempfile
 
 import time
 
 import unittest
 
 from research_session import ResearchSession, ROOT
 
 class ResearchTests(unittest.TestCase):
     def test_prepared_fixture_selections_never_invoke_compiler(self):
         with tempfile.TemporaryDirectory() as directory:
             source = Path(directory) / 'fixture.tri'
             source.write_text('!export *\none = t\ntwo = t t\n')
             session = ResearchSession.__new__(ResearchSession)
             session.records = {source: {'after_sha256': hashlib.sha256(source.read_bytes()).hexdigest(),
                                         'manifest': 'manifest'}}
             session.tricu, session.store = 'tricu', Path(directory)
             session.object = lambda domain, pin: b'header\nexport\tone\tkind\troot1\tabi\nexport\ttwo\tkind\troot2\tabi\n'
             calls = []
             session.run = lambda command, seconds: (calls.append(command) or b'', .01)
             for name in ['one', 'two', 'one']:
                 session.compile(source, [name], Path(directory) / (name + '.arboricx'))
             self.assertEqual(len(calls), 3)
             self.assertTrue(all(command[1:3] == ['arboricx', 'export'] for command in calls))
 
     def test_changed_prepared_source_is_rejected(self):
         with tempfile.TemporaryDirectory() as directory:
             source = Path(directory) / 'fixture.tri'; source.write_text('changed')
             session = ResearchSession.__new__(ResearchSession)
             session.records = {source: {'after_sha256': '0' * 64}}
             with self.assertRaisesRegex(ValueError, 'changed during session'):
                 session.compile(source, ['main'], Path(directory) / 'out')
 
     def test_job_above_twenty_minutes_is_rejected(self):
         session = ResearchSession.__new__(ResearchSession)
         with self.assertRaisesRegex(ValueError, '1200'):
             session.run(['never-started'], 1201)
 
     def test_expired_whole_harness_cannot_start_another_child(self):
         session = ResearchSession.__new__(ResearchSession)
         session.deadline = time.monotonic() - 1
         with self.assertRaises(subprocess.TimeoutExpired):
             session.run(['never-started'], 1200)
 
     def test_ir_harness_uses_original_authenticated_fixture_path(self):
         text = (ROOT / 'test/staging_experiment.py').read_text()
         self.assertIn("source = ROOT / 'test/stagingExperiment.tri'", text)
         self.assertNotIn("source = root / 'fixtures.tri'", text)
         self.assertNotIn('zig', text.lower())
 
     def test_conditional_fixtures_are_batched(self):
         text = (ROOT / 'test/staged_conditional.py').read_text()
         body = text.split('    def fixture_for(', 1)[1].split('    for entry in ', 1)[0]
         self.assertNotIn('compile(', body)
         self.assertIn('arboricx', body)
 
 
 # standard_native_migration_check
 from pathlib import Path
 
 import unittest
 
 ROOT = Path(__file__).resolve().parent.parent
 
 DEFERRED_LIBRARIES = {}
 
 LEGACY_DEMO_FIXTURES = set()
 
 def legacy(path):
     return any(line.startswith('!import "') for line in path.read_text().splitlines())
 
 class StandardMigrationTests(unittest.TestCase):
     def test_remaining_library_imports_are_explicitly_scoped(self):
         remaining = {str(path.relative_to(ROOT)) for path in (ROOT / 'lib').rglob('*.tri')
                      if legacy(path)}
         self.assertEqual(remaining, set(DEFERRED_LIBRARIES))
 
     def test_application_demos_are_native(self):
         demos = [*(ROOT / 'demos').glob('*.tri'), *(ROOT / 'demos/interactionTrees').glob('*.tri')]
         self.assertEqual(len(demos), 23)
         for path in demos:
             with self.subTest(path=path):
                 self.assertFalse(legacy(path))
                 self.assertTrue(any(line.startswith('!export ') for line in path.read_text().splitlines()))
 
     def test_demo_inputs_have_no_legacy_imports(self):
         remaining = {str(path.relative_to(ROOT)) for path in (ROOT / 'demos').rglob('*.tri')
                      if legacy(path)}
         self.assertEqual(remaining, LEGACY_DEMO_FIXTURES)
 
     def test_formatter_and_recursion_are_native(self):
         for directory in ['lib/tools', 'lib/recursion']:
             for path in (ROOT / directory).rglob('*.tri'):
                 with self.subTest(path=path):
                     self.assertFalse(legacy(path))
                     self.assertTrue(any(line.startswith('!export ') for line in path.read_text().splitlines()))
 
 
 class NativeHeaderTests(unittest.TestCase):
     def test_order_and_deduplication(self):
         from native_header import native_header
         edges, imports = native_header("-- header\n!import math\n!export * from base\n!import math as M\n!source build_only\n!export main\nmain = t\n")
         self.assertEqual(edges, ['math', 'base', 'build_only'])
         self.assertEqual(imports, [('math', 'math'), ('math', 'M')])
 
     def test_hash_imports_and_barrels(self):
         from native_header import native_header
         pin = 'a' * 64
         self.assertEqual(native_header('!export * from dep\n')[0], ['dep'])
         self.assertEqual(native_header('!import '+pin+' as M\n!export * from '+pin+'\n')[0], [])
         with self.assertRaises(ValueError): native_header('!import '+pin+'\n')
 
     def test_redundant_build_only_edges(self):
         from native_header import native_header
         for source in ['!source dep\n!import dep\n', '!import dep\n!source dep\n',
                        '!source dep\n!export * from dep\n', '!source dep\n!source dep\n']:
             with self.subTest(source=source), self.assertRaisesRegex(ValueError, 'redundant'):
                 native_header(source)
 
     def test_old_native_syntax_and_recovery_spelling_rejected(self):
         from native_header import native_header
         for source in ['!use dep *\n', '!import "dep" !Local\n']:
             with self.subTest(source=source), self.assertRaises(ValueError): native_header(source)
 
 
 class GuardedJobTests(unittest.TestCase):
     def test_output_and_nonzero_exit(self):
         import subprocess
         import sys
         from process_rss import run_guarded
         output, peak = run_guarded([sys.executable, '-c', 'print("transport-ok")'],
                                   memory_mib=64, seconds=5)
         self.assertEqual(output.strip(), 'transport-ok')
         self.assertGreaterEqual(peak, 0)
         with self.assertRaises(subprocess.CalledProcessError) as failure:
             run_guarded([sys.executable, '-c', 'print("diagnostic"); raise SystemExit(3)'],
                         memory_mib=64, seconds=5)
         self.assertIn('diagnostic', failure.exception.output)
 
     def test_time_and_memory_guards(self):
         import sys
         from process_rss import run_guarded
         with self.assertRaisesRegex(RuntimeError, 'wall-time guard'):
             run_guarded([sys.executable, '-c', 'import time; time.sleep(10)'],
                         memory_mib=64, seconds=0.05)
         with self.assertRaisesRegex(RuntimeError, 'RSS guard'):
             run_guarded([sys.executable, '-c', 'import time; data=bytearray(8*1024*1024); time.sleep(10)'],
                         memory_mib=1, seconds=5)
 
 
 class BootstrapAcceptanceTests(unittest.TestCase):
     def setUp(self):
         from bootstrap_generations import source_map, snapshot
         from bootstrap_acceptance import ARTIFACTS, sha
         self.temporary = tempfile.TemporaryDirectory()
         self.addCleanup(self.temporary.cleanup)
         self.root = Path(self.temporary.name)
         self.source = self.root / 'source'
         self.source.mkdir()
         (self.source / 'root.tri').write_text('!export main\nmain = t\n')
         self.runtime = self.root / 'host'
         self.runtime.write_bytes(b'host-identity')
         self.first, self.second = self.root / 'first', self.root / 'second'
         for index, run in enumerate((self.first, self.second)):
             (run / 'cas/cache/source-build').mkdir(parents=True)
             (run / 'source-map.v1').write_text('arboricx.source-map.v1\nnative\tsource_map\troot\troot.tri\t-\n')
             (run / 'cas/cache/source-build/source_map').write_text('version\tkey\tidentical-manifest\treceipt')
             (run / 'compiler.arboricx').write_bytes(b'bridge' if index == 0 else b'rebuilt-compiler.arboricx')
             (run / 'io-adapter.arboricx').write_bytes(b'adapter')
             for name in ARTIFACTS:
                 (run / name).write_bytes(name.encode())
             text, entries, _ = source_map(run / 'source-map.v1', self.source, 'source_map')
             inputs = dict(source_root=str(self.source), root='source_map', backend='stable', fresh_store=True,
                           runtime_sha256=sha(self.runtime), adapter_sha256=sha(run / 'io-adapter.arboricx'),
                           compiler_bundle_sha256=sha(run / 'compiler.arboricx'), **snapshot(text, entries, self.source))
             (run / 'inputs.json').write_text(json.dumps(inputs))
             (run / 'status.json').write_text(json.dumps(dict(complete=True, **inputs,
                 artifacts={name: dict(sha256=sha(run / name)) for name in ARTIFACTS})))
 
     def verify(self):
         from bootstrap_acceptance import verify
         from unittest.mock import patch
         # Synthetic outputs test gate policy only; real CAS auditing is separate.
         with patch('bootstrap_acceptance.run_guarded', return_value=('synthetic audit', 0)):
             return verify(self.first, self.second, self.runtime)
 
     def test_accepts_matching_policy_records(self):
         self.assertTrue(self.verify()['accepted'])
 
     def test_rejects_wrong_compiler_and_artifact(self):
         from bootstrap_acceptance import sha
         path = self.second / 'compiler.arboricx'
         path.write_bytes(b'unrelated-worker')
         for filename in ('inputs.json', 'status.json'):
             record_path = self.second / filename
             record = json.loads(record_path.read_text())
             record['compiler_bundle_sha256'] = sha(path)
             record_path.write_text(json.dumps(record))
         with self.assertRaisesRegex(ValueError, 'did not execute'):
             self.verify()
         path.write_bytes(b'rebuilt-compiler.arboricx')
         (self.first / 'build-stable.arboricx').write_bytes(b'corrupt')
         with self.assertRaisesRegex(ValueError, 'artifact identity'):
             self.verify()
 
     def test_rejects_seeded_or_failed_runs(self):
         for field, value, diagnostic in [('fresh_store', False, 'fresh-store'),
                                           ('complete', False, 'completed stable')]:
             path = self.second / 'status.json'
             original = path.read_bytes()
             data = json.loads(original)
             data[field] = value
             path.write_text(json.dumps(data))
             with self.assertRaisesRegex(ValueError, diagnostic):
                 self.verify()
             path.write_bytes(original)
 
     def test_rejects_changed_source_and_actual_manifests(self):
         path = self.source / 'root.tri'
         original = path.read_bytes()
         path.write_bytes(original + b'-- changed\n')
         with self.assertRaisesRegex(ValueError, 'fingerprint'):
             self.verify()
         path.write_bytes(original)
         (self.second / 'cas/cache/source-build/source_map').write_text('version\tkey\tdifferent-manifest\treceipt')
         with self.assertRaisesRegex(ValueError, 'manifest reproduction'):
             self.verify()
 
 
 class SourceMappedKitTests(unittest.TestCase):
     def test_projection_keeps_authored_sources_and_explicit_framing(self):
         from prepare_compiler_kit import project
         _, sources, worker, adapter = project(ROOT/'bootstrap/compiler-source-map.v1', ROOT, 2, 'stable')
         self.assertEqual(sources['compiler_io'], (ROOT/'lib/source/io.tri').read_bytes())
         self.assertIn('sourceMapBuildStable = M.sourceMapBuildStable', worker)
         self.assertIn('!import io as IO', adapter)
         self.assertEqual(sources['source_object'], (ROOT/'lib/source/object.tri').read_bytes())
         with self.assertRaisesRegex(ValueError, 'requires IO protocol 2'):
             project(ROOT/'bootstrap/compiler-source-map.v1', ROOT, 1, 'stable')
         self.assertEqual(sources['io'], (ROOT/'lib/io.tri').read_bytes())
         self.assertIn('!import compiler_io *', sources['source_object'].decode())
         _, _, worker, adapter = project(ROOT/'bootstrap/compiler-source-map.v1', ROOT, 2, 'optimized')
         self.assertIn('sourceMapBuild = M.sourceMapBuild', worker)
         self.assertIn('!import io as IO', adapter)
 
 
 class ArchiveRecoveryPolicyTests(unittest.TestCase):
     def test_recovery_binds_produced_version_and_adapter(self):
         from bootstrap_archive import recovery_profile
         import hashlib
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory); adapter = root/'adapter'
             adapter.write_bytes(b'v2')
             pin = hashlib.sha256(b'v2').hexdigest()
             (root/'produced-io-adapter.arboricx').write_bytes(b'v2')
             inputs = dict(execution_io_version=1, produced_io_version=2,
                           produced_adapter_sha256=pin, adapter_sha256='0'*64)
             self.assertEqual(recovery_profile(root, inputs, adapter, True), 2)
             with self.assertRaisesRegex(ValueError, 'accepted pair'):
                 recovery_profile(root, inputs, adapter, False)
             adapter.write_bytes(b'wrong')
             with self.assertRaisesRegex(ValueError, 'produced adapter'):
                 recovery_profile(root, inputs, adapter, True)
             adapter.write_bytes(b'v2')
             (root/'produced-io-adapter.arboricx').write_bytes(b'corrupt')
             with self.assertRaisesRegex(ValueError, 'identity mismatch'):
                 recovery_profile(root, inputs, adapter, True)
             del inputs['execution_io_version']
             with self.assertRaisesRegex(ValueError, 'incomplete'):
                 recovery_profile(root, inputs, adapter, True)
             self.assertEqual(recovery_profile(root, {}, adapter, False), 1)
             with self.assertRaisesRegex(ValueError, 'accepted output'):
                 recovery_profile(root, {'adapter_sha256':'0'*64}, adapter, True)
 
     def test_scp_execution_is_not_a_compiler_restore_gate(self):
         archive = (ROOT / 'test/bootstrap_archive.py').read_text()
         self.assertNotIn('scp_native_validation.py', archive)
         self.assertIn('scp_execution_required=False', archive)
 
 
 class SupportProjectionTests(unittest.TestCase):
     def setUp(self):
         from bundle_dag import merge_bundles, read
         self.temporary = tempfile.TemporaryDirectory()
         self.addCleanup(self.temporary.cleanup)
         self.directory = Path(self.temporary.name)
         self.original = self.directory / 'original.arboricx'
         self.projected = self.directory / 'projected.arboricx'
         leaf = ROOT / 'test/fixtures/false.arboricx'
         other = ROOT / 'test/fixtures/true.arboricx'
         merge_bundles(self.original, [(leaf, [(next(iter(read(leaf)[3])), 'keep.value')]),
                                      (other, [(next(iter(read(other)[3])), 'drop.value')])])
         merge_bundles(self.projected, [(self.original, [('keep.value', 'keep.value')])],
                       prune_unreachable=True)
 
     def test_exact_selected_exports_without_unreachable_nodes(self):
         from project_support_fixture import verify_projection
         self.assertEqual(verify_projection(self.original, self.projected, {'keep'}), 1)
 
     def test_missing_export_and_changed_kernel_rejected(self):
         from bundle_dag import merge_bundles
         from project_support_fixture import verify_projection
         with self.assertRaisesRegex(ValueError, 'export set'):
             verify_projection(self.original, self.projected, {'keep', 'drop'})
         merge_bundles(self.projected, [(self.original, [('drop.value', 'keep.value')])],
                       prune_unreachable=True)
         with self.assertRaisesRegex(ValueError, 'changed a kernel'):
             verify_projection(self.original, self.projected, {'keep'})
 
     def test_unreachable_archived_nodes_rejected(self):
         from bundle_dag import merge_bundles
         from project_support_fixture import verify_projection
         merge_bundles(self.projected, [(self.original, [('keep.value', 'keep.value')])])
         with self.assertRaisesRegex(ValueError, 'unreachable'):
             verify_projection(self.original, self.projected, {'keep'})
 
     def test_immutable_projection_origin_fingerprints(self):
         # The extraction ledger remains historical: current/rebuilt outputs have
         # since been genuinely rebuilt. Never rewrite the original snapshots or
         # pretend new compilation is an unchanged projection. Current outputs
         # and every rebuilt receipt are authenticated by the Haskell cohort gate.
         fixtures = ROOT / 'test/fixtures'
         evidence = json.loads((fixtures / 'support-projection.json').read_text())
         self.assertEqual(evidence['format'], 'tricu.support-projection.v1')
         for name in ['support-origin.v1', 'support-origin.arboricx', 'support-origin-compiler.arboricx']:
             self.assertEqual(hashlib.sha256((fixtures / name).read_bytes()).hexdigest(),
                              evidence['outputs_sha256'][name], name)
 
     def test_active_support_is_dependency_closed_without_archive(self):
         from native_header import native_header
         fixtures = ROOT / 'test/fixtures'
         rows = [line.split('\t') for name in ['compiler-library.v1', 'support-library.v1']
                 for line in (fixtures / name).read_text().splitlines() if line.startswith('module\t')]
         providers = {row[2] for row in rows}
         for _, _, alias, path, pin in rows:
             self.assertNotIn('consumer', path)
             self.assertNotIn('archive/', path)
             from fixture_sources import snapshot_bytes
             dependencies, _ = native_header(snapshot_bytes(fixtures, pin).decode())
             self.assertLessEqual(set(dependencies), providers, alias)
         self.assertNotIn('ScpCommand', (ROOT / 'tricu.cabal').read_text())
         self.assertFalse((ROOT / 'src/ScpCommand.hs').exists())
 
 
 class LibraryPublicationAliases(unittest.TestCase):
     def test_authored_aliases_are_not_filename_guesses(self):
         import importlib.util
         spec = importlib.util.spec_from_file_location('prepare_libraries', ROOT/'ext/demo-site/prepare-libraries.py')
         module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
         rows = module.library_sources("tricu.workspace.v1\nhandler_upload\tlib/handlers/upload.tri\nhandler_testing\tlib/handlers/testing.tri\n")
         self.assertEqual(next(alias for alias,path,_ in rows if path=='lib/handlers/upload.tri'), 'handler_upload')
         self.assertEqual(next(alias for alias,path,_ in rows if path=='lib/handlers/testing.tri'), 'handler_testing')
         self.assertEqual({path for _,path,_ in rows}, {str(path.relative_to(ROOT)) for path in (ROOT/'lib').rglob('*.tri')})
         for text in ['bad', 'tricu.workspace.v1\na\tx\na\ty', 'tricu.workspace.v1\na\tx\nb\tx', 'tricu.workspace.v1\na\t../x']:
             with self.assertRaises(ValueError): module.workspace_aliases(text)
 
 
 class SupportPublicationControls(unittest.TestCase):
     def test_resource_limits_fail_before_build_or_publication(self):
         import contextlib
         import io
         from native_library_migration import main as compile_candidates
         from repair_support_fixture import main as publish_candidates
         cases = [
             (compile_candidates, ['tool', 'host', 'generation', 'output', '--library', 'socket', '--capabilities', '0']),
             (publish_candidates, ['tool', 'host', 'candidates', '--output-directory', 'output', '--capabilities', '0']),
             (publish_candidates, ['tool', 'host', 'candidates', '--output-directory', 'output', '--verification-memory-mib', '65537']),
         ]
         for main, argv in cases:
             with self.subTest(argv=argv), patch('sys.argv', argv), contextlib.redirect_stderr(io.StringIO()):
                 with self.assertRaises(SystemExit) as error:
                     main()
                 self.assertEqual(error.exception.code, 2)
 
     def test_publication_cannot_overwrite_an_existing_directory(self):
         from repair_support_fixture import main
         with tempfile.TemporaryDirectory() as directory:
             marker = Path(directory) / 'marker'
             marker.write_bytes(b'keep')
             with patch('sys.argv', ['tool', 'host', 'candidates', '--output-directory', directory]):
                 with self.assertRaises(FileExistsError):
                     main()
             self.assertEqual(marker.read_bytes(), b'keep')
             self.assertEqual(list(Path(directory).iterdir()), [marker])
 
 
 class CompilerDependencyTests(unittest.TestCase):
     def test_current_map_is_exact_compiler_closure_without_patterns(self):
         from bootstrap_generations import source_map
         _, entries, required = source_map(
             ROOT / 'bootstrap/compiler-source-map.v1', ROOT, 'source_map')
         self.assertEqual(entries['compiler_io'][2], ['base', 'io', 'io_storage', 'io_text'])
         self.assertEqual(entries['io'][1], 'lib/io.tri')
         self.assertNotIn('patterns', entries)
         self.assertEqual(set(entries), required)
 
 
 class FixtureSourceTests(unittest.TestCase):
     def setUp(self):
         from fixture_sources import digest
         self.temporary = tempfile.TemporaryDirectory()
         self.addCleanup(self.temporary.cleanup)
         self.root = Path(self.temporary.name)
         self.fixtures, self.source = self.root / 'fixtures', self.root / 'source'
         self.fixtures.mkdir()
         self.source.mkdir()
         for cohort, alias in [('compiler-library', 'compiler'), ('support-library', 'support')]:
             source = b'!export main\nmain = t\n' + ('-- ' + alias + '\n').encode()
             (self.source / (alias + '.tri')).write_bytes(source)
             bundle = b'synthetic-policy-bundle-' + alias.encode()
             (self.fixtures / (cohort + '.arboricx')).write_bytes(bundle)
             (self.fixtures / (cohort + '.v1')).write_text(
                 'tricu.' + cohort + '-fixture.v1\nmodule\t' + alias + '\t' + alias + '\t' + alias + '.tri\t'
                 + digest('arboricx.source.v1', source) + '\nbundle\t'
                 + digest('tricu.' + cohort + '-fixture.v1', bundle) + '\n')
         (self.source / 'bootstrap').mkdir()
         (self.source / 'bootstrap/compiler-source-map.v1').write_text(
             'arboricx.source-map.v1\nnative\tcompiler\tcompiler\tcompiler.tri\t-\n')
 
     def capture(self):
         from fixture_sources import capture
         return capture(self.fixtures, self.source)
 
     def test_reference_and_freshness_are_distinct(self):
         from fixture_sources import verify, current_mismatches
         self.capture()
         self.assertEqual(current_mismatches(self.fixtures, self.source), [])
         (self.source / 'compiler.tri').write_text('!import changed *\n!export main\nmain = changed\n')
         self.assertEqual(len(verify(self.fixtures)), 2)
         self.assertEqual(current_mismatches(self.fixtures, self.source), ['compiler.tri'])
 
     def test_materialize_authenticates_reference_without_blessing_worktree(self):
         from fixture_sources import materialize, current_mismatches
         self.capture()
         (self.source / 'support.tri').write_bytes(b'changed work')
         target = self.root / 'reference'
         materialize(self.fixtures, target)
         self.assertEqual(current_mismatches(self.fixtures, target), [])
         self.assertEqual(current_mismatches(self.fixtures, self.source), ['support.tri'])
         with self.assertRaises(FileExistsError): materialize(self.fixtures, target)
         next((self.fixtures / 'reference-source').glob('*.tri')).write_bytes(b'corrupt')
         with self.assertRaises(ValueError): materialize(self.fixtures, self.root / 'bad')
         self.assertFalse((self.root / 'bad').exists())
 
     def test_capture_cannot_bless_edited_sources(self):
         (self.source / 'support.tri').write_text('changed')
         with self.assertRaisesRegex(ValueError, 'does not match'):
             self.capture()
         self.assertFalse((self.fixtures / 'reference-source').exists())
 
     def test_missing_snapshot_never_falls_back_to_matching_working_source(self):
         from fixture_sources import verify
         with self.assertRaises(FileNotFoundError):
             verify(self.fixtures)
 
     def test_corrupt_snapshot_is_rejected_not_repaired(self):
         from fixture_sources import verify
         self.capture()
         next((self.fixtures / 'reference-source').glob('*.tri')).write_bytes(b'corrupt')
         with self.assertRaisesRegex(ValueError, 'digest mismatch'):
             verify(self.fixtures)
         with self.assertRaisesRegex(ValueError, 'existing reference'):
             self.capture()
 
     def test_corrupt_bundle_is_rejected(self):
         from fixture_sources import verify
         self.capture()
         (self.fixtures / 'compiler-library.arboricx').write_bytes(b'corrupt')
         with self.assertRaisesRegex(ValueError, 'bundle digest'):
             verify(self.fixtures)
 
     def test_unsafe_or_duplicate_metadata_is_rejected(self):
         from fixture_sources import source_rows
         path = self.fixtures / 'compiler-library.v1'
         original = path.read_text()
         for value in ['../outside', '/outside']:
             path.write_text(original.replace('compiler.tri', value))
             with self.assertRaisesRegex(ValueError, 'path or digest'):
                 source_rows(self.fixtures)
         path.write_text(original + original.splitlines()[1] + '\n')
         with self.assertRaisesRegex(ValueError, 'duplicate'):
             source_rows(self.fixtures)
 
     def test_cli_reports_reference_success_but_current_failure(self):
         import sys
         self.capture()
         (self.source / 'compiler.tri').write_bytes(b'edited')
         base = [sys.executable, str(ROOT / 'test/fixture_sources.py')]
         common = ['--fixtures', str(self.fixtures), '--source-root', str(self.source)]
         reference = subprocess.run(base + ['verify'] + common, capture_output=True, text=True)
         current = subprocess.run(base + ['current'] + common, capture_output=True, text=True)
         self.assertEqual(reference.returncode, 0, reference.stderr)
         self.assertIn('freshness NOT checked', reference.stdout)
         self.assertEqual(current.returncode, 1)
         self.assertIn('compiler.tri', current.stderr)
 
     def test_map_change_is_freshness_failure(self):
         from fixture_sources import current_mismatches
         self.capture()
         (self.source / 'bootstrap/compiler-source-map.v1').write_text('arboricx.source-map.v1\n')
         self.assertEqual(current_mismatches(self.fixtures, self.source), ['bootstrap/compiler-source-map.v1'])
 
 
 class SupportPublicationTests(unittest.TestCase):
     def test_patterns_relocation_is_explicit_and_identity_checked(self):
         from repair_support_fixture import relocate_patterns, updated_module_rows
         old = ['module\thttp\thttp\tlib/http.tri\told-source']
         record = dict(legacy='patterns', path='lib/patterns.tri')
         candidates = {'patterns': (record, b'!export main\nmain = t\n')}
         with self.assertRaisesRegex(ValueError, 'membership'):
             updated_module_rows(old, candidates)
         relocated = relocate_patterns(old, candidates)
         self.assertEqual(updated_module_rows(relocated, candidates), relocated)
         self.assertEqual(old, ['module\thttp\thttp\tlib/http.tri\told-source'])
         with self.assertRaisesRegex(ValueError, 'requires rebuilt'):
             relocate_patterns(old, {})
         with self.assertRaisesRegex(ValueError, 'identity'):
             relocate_patterns(old, {'patterns': (dict(record, path='other.tri'), b't')})
         with self.assertRaisesRegex(ValueError, 'already belongs'):
             relocate_patterns(relocated, candidates)
 
     def test_changed_source_has_a_new_authenticated_fingerprint(self):
         from repair_support_fixture import updated_module_rows
         old = ['module\thttp\thttp\tlib/http.tri\told-source',
                'module\tsocket\tsocket\tlib/socket.tri\toriginal-socket']
         record = dict(legacy='http', path='lib/http.tri')
         payload = b'!export f\nf x = x\n'
         rows = updated_module_rows(old, {'http': (record, payload)})
         self.assertEqual(rows[0].split('\t')[-1], digest('arboricx.source.v1', payload))
         self.assertEqual(rows[1], old[1])
         self.assertEqual(old[0].split('\t')[-1], 'old-source')
         with self.assertRaisesRegex(ValueError, 'identity'):
             updated_module_rows(old, {'http': (dict(record, path='other.tri'), payload)})
         with self.assertRaisesRegex(ValueError, 'membership'):
             updated_module_rows(old, {'unknown': (record, payload)})
 
     def test_candidate_replacement_is_explicit_and_complete(self):
         from repair_support_fixture import select_candidates
         with tempfile.TemporaryDirectory() as directory:
             roots = [Path(directory) / str(n) for n in range(2)]
             for root in roots:
                 root.mkdir()
                 (root / 'migration.json').write_text(json.dumps([dict(alias='http', complete=True)]))
             with self.assertRaisesRegex(ValueError, 'requires --replace'):
                 select_candidates(roots, [])
             self.assertEqual(select_candidates(roots, ['http'])['http'][0], roots[1])
             with self.assertRaisesRegex(ValueError, 'exactly'):
                 select_candidates(roots[:1], ['http'])
             (roots[1] / 'migration.json').write_text(json.dumps([dict(alias='http', complete=False)]))
             with self.assertRaisesRegex(ValueError, 'incomplete'):
                 select_candidates(roots, ['http'])
 
     def test_receipt_cannot_relabel_source_compiler_or_output(self):
         from repair_support_fixture import checked_receipt
         source = b'f x = x\n'
         payload = ('arboricx.build-receipt.v1\nsource\t' + digest('arboricx.source.v1', source) +
                    '\ncompiler\tcompiler\noutput\tmanifest\noptions\t' +
                    digest('arboricx.build-options.v1', b'source-module.stable.v1') + '\n').encode()
         checked_receipt(payload, source, 'manifest', 'compiler')
         for raw, text, manifest, compiler in [
             (payload, source + b'-- changed\n', 'manifest', 'compiler'),
             (payload, source, 'wrong', 'compiler'),
             (payload, source, 'manifest', 'wrong'),
             (payload + b'output\tmanifest\n', source, 'manifest', 'compiler'),
             (payload.replace(digest('arboricx.build-options.v1', b'source-module.stable.v1').encode(),
                              digest('arboricx.build-options.v1', b'source-module.reference.v1').encode()),
              source, 'manifest', 'compiler'),
         ]:
             with self.subTest(manifest=manifest, compiler=compiler), self.assertRaisesRegex(ValueError, 'receipt mismatch'):
                 checked_receipt(raw, text, manifest, compiler)
 
     def test_old_consumer_receipt_cannot_pin_an_old_support_provider(self):
         from repair_support_fixture import check_support_dependencies
         providers = dict(http='current-http', socket='current-socket')
         rows = [['local', 'http', 'current-http', 'facade-source'],
                 ['pin', 'http', 'current-http', 'parseHeaders', 'tree']]
         check_support_dependencies(rows, providers)
         for bad in ([['local', 'http', 'old-http', 'source']],
                     [rows[0], ['pin', 'http', 'old-http', 'parseHeaders', 'tree']],
                     [rows[0], rows[0]], [['local', 'http']], [['pin', 'http']]):
             with self.subTest(bad=bad), self.assertRaises(ValueError):
                 check_support_dependencies(bad, providers)
 
 
 class LocalControlHarnessTests(unittest.TestCase):
     def test_independent_case_diagnostic_and_ast_corpus(self):
         from local_control import cases, failures, ast_cases
         rows = cases()+failures()+ast_cases()
         self.assertEqual(len(rows),len({row[0] for row in rows}))
         labels = {row[0] for row in rows}
         for label in ('case-ignored-scrutinee', 'case-unused-loop', 'case-pin-collision',
                       'rec-nested-capture', 'rec-no-mutual', 'plain-let-shadow',
                       'rec-value-before-missing', 'case-ast', 'rec-ast'):
             self.assertIn(label,labels)
         from http_headers import expect
         for _,_,expected in ast_cases():
             with self.assertRaises(ValueError):
                 expect(expected+b'\0', expected, 'extra AST field')
 
     def test_explicit_source_dependencies_cannot_override_or_escape(self):
         from native_library_migration import dependency_sources, check_source_dependencies
         import tempfile, json
         from pathlib import Path
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory); (root/'extra.tri').write_text('!export main\nmain = 42\n')
             rows = dependency_sources(root, ['extra=extra.tri'], {'base'})
             workspace = root/'workspace'; workspace.mkdir()
             (workspace/'extra.tri').write_bytes((root/'extra.tri').read_bytes())
             (root/'selection.json').write_text(json.dumps(dict(source_dependencies=rows)))
             check_source_dependencies(root, root)
             for entries in [['base=extra.tri'], ['extra=extra.tri']*2, ['extra=../escape'], ['extra=/absolute'], ['!bad=extra.tri'], ['extra']]:
                 with self.assertRaises(ValueError): dependency_sources(root, entries, {'base'})
             (root/'outside.tri').symlink_to('/etc/passwd')
             with self.assertRaises(ValueError): dependency_sources(root, ['extra=outside.tri'], set())
             (root/'extra.tri').write_text('!export main\nmain = 43\n')
             with self.assertRaises(ValueError): check_source_dependencies(root, root)
 
     def test_candidate_source_guard_rejects_old_facade_and_source_edits(self):
         from native_library_migration import checked_candidate_source
         import tempfile, hashlib
         from pathlib import Path
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory); workspace = root/'workspace'; workspace.mkdir()
             source = b'!export value\nvalue = 23\n'
             (root/'library.tri').write_bytes(source)
             staged = workspace/'source_ast.tri'; staged.write_bytes(source)
             record = dict(path='library.tri', alias='source_ast', after_sha256=hashlib.sha256(source).hexdigest())
             self.assertEqual(checked_candidate_source(workspace, record, root), source)
             staged.write_bytes(b'!export * from '+b'0'*64+b'\n')
             with self.assertRaises(ValueError): checked_candidate_source(workspace, record, root)
             staged.write_bytes(source); (root/'library.tri').write_bytes(source+b'-- changed\n')
             with self.assertRaises(ValueError): checked_candidate_source(workspace, record, root)
 
 
 class RegistryServerHarnessTests(unittest.TestCase):
     def test_independent_oracles_and_explicit_changes(self):
         from registry_server import pure_cases, effect_cases, demo_cases
         rows = pure_cases() + effect_cases() + list(demo_cases())
         self.assertEqual(len(rows), len({row.label for row in rows}))
         self.assertEqual(sum(c.label.startswith('hex-byte-') for c in rows), 256)
         self.assertEqual(sum(c.label.startswith('hash-invalid-at-') for c in rows), 64)
         self.assertEqual(sum(c.label.startswith('demo-init-failure-') for c in rows), 7)
         for c in rows:
             self.assertIsInstance(c.expected, bytes)
             self.assertEqual(bool(c.change), c.previous is not None)
             if c.previous is not None:
                 self.assertNotEqual(c.expected, c.previous)
 
     def test_traversal_never_requests_filesystem_io(self):
         import posixpath
         from registry_server import effect_cases, finished, result, number, bs, DIAGNOSTIC, action, ROOT, path
         rows = effect_cases()
         for c in rows:
             if c.label.startswith('get-reject-'):
                 self.assertEqual(c.expected, finished(result(False, number(400), bs(DIAGNOSTIC))))
             if c.label.startswith('write-reject-'):
                 self.assertEqual(c.expected, finished(result(False, bs(b'sha256Hex: '+DIAGNOSTIC)),
                                                       [action(28, bs(b'body'))]))
                 self.assertEqual(c.previous_export, 'putBundleWithHash')
         self.assertFalse(posixpath.normpath(path(b'../../../escaped')).startswith(ROOT+b'/'))
 
     def test_prior_defects_cannot_pass_current_oracles(self):
         from registry_server import effect_cases, demo_cases
         from http_pipeline import check_case
         for c in effect_cases()+list(demo_cases()):
             if c.previous is not None:
                 check_case(c, c.expected, c.previous)
                 with self.assertRaises(ValueError):
                     check_case(c, c.previous, c.previous)
 
     def test_wrong_path_and_atomic_operation_are_detected(self):
         from registry_server import action, bs, pair, finished, result, expect
         output = result(True, bs(b'hash'))
         write = action(33, pair(bs(b'/registry/objects/abc/hash.arboricx'), bs(b'data')))
         expected = finished(output, [write])
         for trace in ([], [action(22, pair(bs(b'/registry/tmp/hash.tmp'), bs(b'data')))],
                       [action(33, pair(bs(b'/outside'), bs(b'data')))], [write, write]):
             with self.assertRaises(ValueError):
                 expect(finished(output, trace), expected, 'mutated publication')
 
 
 class HttpPipelineHarnessTests(unittest.TestCase):
     def test_changes_have_explicit_old_and_independent_new_results(self):
         from http_pipeline import pure_cases, effect_cases
         rows = pure_cases() + effect_cases()
         self.assertEqual(len(rows), len({row.label for row in rows}))
         changes = {row.label for row in rows if row.previous is not None}
         self.assertEqual(changes, {'body-eof-', 'body-eof-6162', 'receive-recv-failure',
             'handler-pure-too-large', 'handler-io-too-large', 'handler-pure-header-limit',
             'handler-io-header-limit', 'handler-pure-body-eof', 'handler-io-body-eof', 'status-413'})
         for row in rows:
             self.assertIsInstance(row.expected, bytes)
             if row.previous is not None:
                 self.assertNotEqual(row.expected, row.previous)
 
     def test_reference_modes_never_disable_subject_oracle(self):
         from http_pipeline import Case, check_case
         row = Case('changed', 'probe', (), b'\0', previous=b'\1\0')
         check_case(row, row.expected, row.previous)
         check_case(row, row.expected, row.expected, reference_current=True)
         for actual, before, current in [(row.previous, row.previous, False),
                                         (row.previous, row.expected, True),
                                         (row.expected, row.previous, True),
                                         (row.expected, row.expected, False)]:
             with self.assertRaises(ValueError):
                 check_case(row, actual, before, current)
 
     def test_both_router_paths_have_success_failure_and_cleanup_cases(self):
         from http_pipeline import effect_cases
         labels = {row.label for row in effect_cases()}
         for flavor in ('pure', 'io'):
             for suffix in ('get', 'post', 'head', 'bad-line', 'bad-header-head', 'bad-method',
                            'bad-version-head', 'bad-target-head', 'duplicate', 'coding-head',
                            'too-large', 'header-eof', 'header-limit', 'body-fragmented',
                            'body-recv-error', 'body-eof', 'body-send-error', 'body-close-error'):
                 self.assertIn('handler-' + flavor + '-' + suffix, labels)
         for control in ('pure', 'extra-script', 'unscripted', 'wrong-tag', 'abort', 'bind', 'fuel'):
             self.assertIn('driver-' + control, labels)
 
     def test_trace_oracle_detects_missing_reordered_or_wrong_effect_payloads(self):
         from http_pipeline import action, finished, number, byte_tree, pair, result, LEAF
         from http_headers import expect
         send = action(77, pair(number(7), byte_tree(b'response')))
         close = action(71, number(7))
         expected = finished(result(True, LEAF), [send, close])
         for trace in ([send], [close, send], [send, close, close],
                       [action(77, pair(number(7), byte_tree(b'wrong'))), close]):
             with self.assertRaises(ValueError):
                 expect(finished(result(True, LEAF), trace), expected, 'trace mutation')
 
 
 class HttpLengthHarnessTests(unittest.TestCase):
     def test_length_corpus_covers_bytes_and_policies_without_running_artifacts(self):
         from http_lengths import cases, BAD_REQUEST
         rows = cases(include_costs=False)
         self.assertEqual(len(rows), len({row.label for row in rows}))
         self.assertEqual(sum(row.label.startswith('value-byte-') for row in rows), 256)
         self.assertEqual(sum(row.label.startswith('duplicate-') for row in rows), 25)
         for row in rows:
             if row.label.startswith(('duplicate-', 'coding-')):
                 self.assertEqual(row.expected, BAD_REQUEST)
                 self.assertFalse(row.compare_reference)
             self.assertEqual(len(row.arguments), 3 if row.export == 'readBody' else 1)
 
     def test_rejections_require_complete_pure_action_not_just_a_status(self):
         from http_lengths import cases, pair, number, BAD_REQUEST, TOO_LARGE
         from http_headers import expect
         rows = cases(include_costs=False)
         by_name = {row.label: row for row in rows}
         for row in rows:
             if row.label.startswith('no-io-'):
                 original = by_name[row.label[len('no-io-'):]]
                 self.assertIn(original.expected, (BAD_REQUEST, TOO_LARGE))
                 self.assertEqual(row.expected, pair(number(0), original.expected))
                 with self.assertRaises(ValueError):
                     expect(pair(number(1), original.expected), row.expected, row.label)
 
     def test_reference_comparisons_never_feed_large_valid_numbers_to_old_parser(self):
         from http_lengths import cases, length_result, BAD_REQUEST, pair, number, result, byte_tree, LEAF
         safe_results = {BAD_REQUEST, length_result(None), length_result(0), length_result(1), length_result(42),
                         pair(number(0), result(True, LEAF, byte_tree(b'body')))}
         for row in cases(include_costs=False):
             if row.compare_reference:
                 self.assertIn(row.expected, safe_results, row.label)
 
 
 class HttpHeaderHarnessTests(unittest.TestCase):
     """Cheap oracle/identity controls only; artifact execution stays opt-in."""
 
     def test_syntax_changes_have_explicit_reasons_and_independent_oracles(self):
         from http_headers import cases, syntax_cases
         old = {case.label: case for case in cases()}
         rows = syntax_cases()
         self.assertEqual(len(rows), len({row.label for row in rows}))
         self.assertEqual(sum(row.label.startswith('name-byte-') for row in rows), 256)
         self.assertEqual(sum(row.label.startswith('value-byte-') for row in rows), 256)
         self.assertTrue(any(row.export == 'parseRequestLine' for row in rows))
         for row in rows:
             if row.change:
                 self.assertIsNotNone(row.expected, row.label)
                 if row.label in old:
                     self.assertEqual(row.reference_expected, old[row.label].expected)
             elif row.label in old:
                 self.assertEqual(row, old[row.label])
 
     def test_only_approved_nonbyte_cases_are_retired(self):
         from http_headers import cases, syntax_cases, retired_case
         historical = cases()
         current = {case.label for case in syntax_cases()}
         retired = {case.label for case in historical if retired_case(case)}
         self.assertEqual(len(retired), 103)
         self.assertEqual({case.label for case in historical} - current, retired)
         self.assertEqual(sum(case.label.startswith('ascii-') for case in syntax_cases()), 256)
         self.assertIn('headers-shape-5', current)
         self.assertIn('headers-shape-6', current)
 
     def test_changed_expectations_do_not_disable_result_or_reference_checks(self):
         from dataclasses import replace
         from http_headers import syntax_cases, check_case, LEAF
         changed = next(row for row in syntax_cases() if row.label == 'bare-lf')
         check_case(changed, changed.reference_expected, changed.expected)
         for before, after in [(changed.reference_expected, changed.reference_expected),
                               (LEAF, changed.expected)]:
             with self.assertRaisesRegex(ValueError, 'complete result differs'):
                 check_case(changed, before, after)
         with self.assertRaisesRegex(ValueError, 'independent oracle'):
             check_case(replace(changed, expected=None), LEAF, LEAF)
         unchanged = next(row for row in syntax_cases() if row.label == 'headers-shape-5')
         with self.assertRaisesRegex(ValueError, 'differential'):
             check_case(unchanged, LEAF, b'\x01\x00')
         with self.assertRaisesRegex(ValueError, 'independent oracle'):
             check_case(unchanged, None, LEAF)
 
     def test_independent_tree_encodings(self):
         from http_headers import number, pair, tree_list, byte_tree, parse_tree_output
         self.assertEqual(number(0), b'\x00')
         self.assertEqual(number(1), b'\x02\x01\x00\x00')
         self.assertEqual(number(2), b'\x02\x00\x02\x01\x00\x00')
         self.assertEqual(pair(b'\x00', b'\x01\x00'), b'\x02\x00\x01\x00')
         self.assertEqual(tree_list([number(65)]), byte_tree(b'A'))
         self.assertEqual(parse_tree_output(' 2010\n'), b'\x02\x00\x01\x00')
 
     def test_renderer_rejects_partial_or_extra_output(self):
         from http_headers import parse_tree_output
         for text in ('', '1', '20', '00', '0 error', '3', '0\n0'):
             with self.subTest(text=text), self.assertRaises(ValueError):
                 parse_tree_output(text)
 
     def test_corpus_is_named_and_policy_characterization_is_explicit(self):
         from http_headers import cases, EXPORTS
         rows = cases()
         self.assertEqual(len(rows), len({row.label for row in rows}))
         self.assertEqual({row.export for row in rows}, set(EXPORTS))
         self.assertEqual(sum(row.label.startswith('ascii-') for row in rows), 256)
         for row in rows:
             self.assertEqual(row.expected is None, row.category == 'differential-only')
         by_name = {row.label: row for row in rows}
         for label in ('embedded-cr', 'bare-lf', 'space-name', 'eof-header', 'duplicates'):
             # Duplicate preservation is ordinary parser behavior; deciding how
             # particular headers interact belongs to a later semantic slice.
             expected = 'independent' if label == 'duplicates' else 'characterization'
             self.assertEqual(by_name[label].category, expected)
 
     def test_complete_result_comparison_rejects_wrong_fields(self):
         from http_headers import cases, expect, headers, reply, diagnostic
         by_name = {row.label: row for row in cases()}
         wrong = [
             ('ascii-065', by_name['ascii-064'].expected),
             ('blank', headers([], b'wrong remainder')),
             ('reply-missing-colon', reply(False, diagnostic(1, b'colon'), consumed=False)),
             ('reply-missing-colon', reply(False, diagnostic(2, b'colon'))),
         ]
         for label, value in wrong:
             with self.subTest(label=label), self.assertRaisesRegex(ValueError, 'complete result differs'):
                 expect(by_name[label].expected, value, label)
 
     def test_reference_integrity_and_recorded_source(self):
         from http_headers import reference, ROOT
         bundle, metadata = reference(ROOT / 'test/fixtures')
         self.assertTrue(bundle.is_file())
         self.assertEqual(metadata['source_path'], 'lib/http.tri')
 
     def test_corrupt_reference_metadata_bundle_and_source_fail(self):
         import shutil
         from http_headers import reference, ROOT
         fixtures = ROOT / 'test/fixtures'
         metadata = json.loads((fixtures / 'http-headers-reference.json').read_text())
         paths = ['http-headers-reference.json', 'http-headers-reference.arboricx',
                  'reference-source/' + metadata['source_digest'] + '.tri']
         for damaged in paths:
             with self.subTest(damaged=damaged), tempfile.TemporaryDirectory() as directory:
                 destination = Path(directory)
                 for path in paths:
                     (destination / path).parent.mkdir(parents=True, exist_ok=True)
                     shutil.copyfile(fixtures / path, destination / path)
                 with (destination / damaged).open('ab') as output:
                     output.write(b'corrupt')
                 with self.assertRaises(ValueError):
                     reference(destination)
 
 
 class ArboricxReaderOracleTests(unittest.TestCase):
     def test_cases_have_unique_names_and_independent_truncation_oracles(self):
         from arboricx_readers import cases
         rows = list(cases())
         self.assertEqual(len(rows), 401)
         self.assertEqual(len({r[1] for r in rows}), len(rows))
         for module in ['common', 'nodes', 'manifest']:
             self.assertTrue(any(r[0] == module and '-cut-' in r[1] and r[4] is not None for r in rows))
 
     def test_binary_failure_keeps_field_start_not_consumed_prefix(self):
         from arboricx_readers import Reader, Failure
         from http_headers import result, number
         from stable_bundle import byte_tree
         r = Reader(b'xx')
         r.take(1)
         with self.assertRaises(Failure) as caught:
             r.take(2)
         self.assertEqual(caught.exception.tree, result(False, number(1), byte_tree(b'x')))
 
 
 class SocketConvenienceOracleTests(unittest.TestCase):
     def test_only_post_acquisition_failures_change(self):
         from socket_conveniences import cases
         rows = list(cases())
         self.assertEqual(len(rows), 14)
         self.assertEqual(len({r.label for r in rows}), len(rows))
         changed = [r for r in rows if r.previous is not None]
         self.assertEqual(len(changed), 6)
         self.assertTrue(all('failure-close-' in r.label and r.previous != r.expected for r in changed))
 
 
 class ParserControlOracleTests(unittest.TestCase):
     def test_complete_reply_and_poison_probe_inventory(self):
         from parser_control import cases
         rows = list(cases())
         self.assertEqual(len(rows), 252)
         self.assertEqual(len({r[0] for r in rows}), len(rows))
         self.assertEqual(sum('ignored-fields-' in r[0] for r in rows), 4)
         self.assertTrue(any(r[0] == 'unselected-failure' for r in rows))
         self.assertTrue(any(r[0] == 'unselected-success' for r in rows))
 
 
 
 class BaseWorkloadTests(unittest.TestCase):
     def test_generated_worker_variants(self):
         from base_workloads import candidate, WORKERS, reference_source
         source = reference_source()
         for variant in ['manual', 'direct', 'patterned', 'captured']:
             text = candidate(source, variant)
             for public in WORKERS:
                 self.assertIn(public + ' ', text)
             if variant != 'manual':
                 self.assertNotRegex(text, r'\bself\b')
         self.assertIn('map f l = (let rec', candidate(source, 'captured'))
         self.assertIn('foldl f x l = (let rec', candidate(source, 'captured'))
         self.assertIn('reverse_ other acc = other', candidate(source, 'patterned'))
 
     def test_large_rendered_result_before_rts_residency(self):
         from base_workloads import measurements
         with tempfile.TemporaryDirectory() as d:
             path = Path(d) / 'sample.log'
             rows = [dict(profile=name, wall_ns=i * 100, allocated_bytes=i * 10,
                          gc_elapsed_ns=i, live_bytes=7)
                     for i, name in enumerate(['load-begin', 'load-end',
                                               'invoke-begin-1', 'invoke-end-1',
                                               'observe-begin-1', 'observe-end-1'])]
             # Must not perform an unanchored numeric regex search over this line.
             path.write_text('\n'.join(json.dumps(r) for r in rows) + '\n' +
                             '0' * 200000 + ' 9,999 bytes allocated in the heap\n' +
                             '   1,234 bytes maximum residency (2 samples)\n')
             actual = measurements(path)
             self.assertEqual(actual['rts_max_residency_bytes'], 1234)
             self.assertEqual(actual['invoke_wall_ns'], 300)
             self.assertEqual(actual['invoke_allocated_bytes'], 30)
             self.assertEqual(actual['measurement_scope'], 'invocation-through-rendered-output-v1')
 
     def test_workload_independent_oracles(self):
         from base_workloads import workloads, encode
         rows = {name: (argument, expected) for name, argument, expected in workloads(1024)}
         data = bytes(rows['lower'][0])
         self.assertEqual(bytes(rows['lower'][1]), data.lower())
         self.assertEqual(rows['splitLines'][1], [list(x) for x in data.split(b'\n')])
         self.assertEqual(rows['sumRows'][1], [sum(x) for x in rows['sumRows'][0]])
         self.assertEqual(len(rows['prefixDigits'][1]), 20)
         self.assertEqual(encode([0, 1]), b'\2\0\2\2\1\0\0\0')
 
     def test_summary_discards_warmup(self):
         from base_workloads import summarize
         fields = ['invoke_wall_ns', 'invoke_allocated_bytes', 'invoke_gc_elapsed_ns',
                   'load_wall_ns', 'load_allocated_bytes', 'post_invoke_live_bytes',
                   'rts_max_residency_bytes', 'peak_kib', 'seconds']
         rows = [dict(variant='manual', workload='lower', size=64, warmup=(v == 999),
                      **{k: v for k in fields}) for v in [999, 2, 4, 8]]
         result = summarize(rows)[0]
         self.assertEqual(result['repetitions'], 3)
         self.assertEqual(result['invoke_wall_ns'], dict(median=4, minimum=2, maximum=8))
 
 
 
 class BaseReadableOracleTests(unittest.TestCase):
     def test_small_bounded_inventory(self):
         from base_readable import cases, edges
         from source_order import end
         rows = list(edges())
         self.assertEqual(len(rows), 30)
         self.assertEqual(len({row[0] for row in rows}), 30)
         for _, _, argument, expected in rows:
             self.assertEqual(end(argument), len(argument))
             self.assertEqual(end(expected), len(expected))
         for size in [64, 1024]:
             workloads = list(cases(size))
             self.assertEqual(len(workloads), 8)
             for _, argument, expected in workloads:
                 self.assertEqual(end(argument), len(argument))
                 self.assertEqual(end(expected), len(expected))
 
 
 class CollectionBatchTests(unittest.TestCase):
     def test_oracles_are_complete_trees_with_unique_short_labels(self):
         from collections_batch import cases, effects
         from source_order import end
         rows = cases() + effects()
         self.assertEqual(len({row[0] for row in rows}), len(rows))
         for label, _, argument, expected in rows:
             self.assertLess(len(label), 120)
             self.assertEqual(end(argument), len(argument))
             self.assertEqual(end(expected), len(expected))
 
     def test_workspace_audit_rejects_changed_sources_and_objects(self):
         from collections_batch import audit_workspace
         with tempfile.TemporaryDirectory() as directory:
             workspace, store, compiler, tree, restore = self.make_workspace(Path(directory))
             self.assertEqual(set(audit_workspace(workspace, store, compiler)), {'a', 'b'})
             source = workspace/'a.tri'
             source.write_text('!export value\nvalue = t t\n')
             with self.assertRaisesRegex(ValueError, 'source'):
                 audit_workspace(workspace, store, compiler)
             source.write_text('!export value\nvalue = t\n')
             (store/'objects'/tree[:3]/tree).write_bytes(b'corrupt')
             with self.assertRaisesRegex(AssertionError, 'object digest'):
                 audit_workspace(workspace, store, compiler)
 
     def test_workspace_audit_rejects_stale_dependency_and_export_pins(self):
         from collections_batch import audit_workspace
         with tempfile.TemporaryDirectory() as directory:
             workspace, store, compiler, tree, restore = self.make_workspace(Path(directory))
             restore(stale_receipt=True)
             with self.assertRaisesRegex(AssertionError, 'stale candidate dependency'):
                 audit_workspace(workspace, store, compiler)
             restore(stale_pin=True)
             with self.assertRaisesRegex(AssertionError, 'stale candidate export pin'):
                 audit_workspace(workspace, store, compiler)
 
     @staticmethod
     def make_workspace(root):
         workspace, store = root/'workspace', root/'store'
         workspace.mkdir();(store/'cache/source-build').mkdir(parents=True)
         compiler = 'a'*64
         def obj(domain, payload):
             pin = digest(domain, payload)
             path = store/'objects'/pin[:3]/pin;path.parent.mkdir(parents=True, exist_ok=True)
             path.write_bytes(payload);return pin
         tree = obj('arboricx.tree-term.v1', b'\0')
         sources = {'a': '!export value\nvalue = t\n',
                    'b': '!import a *\n!export main\nmain = value\n'}
         manifests = {alias: obj('arboricx.module-manifest.v1',
             ('arboricx.module-manifest.v1\nexport\t'+name+'\tarboricx.tree-term.v1\t'+tree+'\tarboricx.abi.tree.v1\n').encode())
             for alias, name in [('a', 'value'), ('b', 'main')]}
         for alias, source in sources.items():
             (workspace/(alias+'.tri')).write_text(source)
         def receipt(alias, extra=''):
             payload = ('arboricx.build-receipt.v1\nsource\t'+digest('arboricx.source.v1', sources[alias].encode())+
                 '\ncompiler\t'+compiler+'\noptions\t'+digest('arboricx.build-options.v1', b'source-module.stable.v1')+
                 '\noutput\t'+manifests[alias]+'\n'+extra).encode()
             pin = obj('arboricx.build-receipt.v1', payload)
             (store/'cache/source-build'/alias).write_text('arboricx.source-build-cache.v1\t'+'0'*64+'\t'+manifests[alias]+'\t'+pin)
             return pin
         a_receipt = receipt('a')
         def restore(stale_receipt=False, stale_pin=False):
             receipt('b', 'local\ta\t'+manifests['a']+'\t'+('b'*64 if stale_receipt else a_receipt)+
                 '\npin\ta\t'+manifests['a']+'\tvalue\t'+('c'*64 if stale_pin else tree)+'\n')
         restore()
         return workspace, store, compiler, tree, restore
 
 
 class NumericBatchTests(unittest.TestCase):
     def test_independent_numeric_oracles_are_complete_trees(self):
         from numeric_batch import cases
         from source_order import end
         rows = cases()
         self.assertEqual(len({row[0] for row in rows}), len(rows))
         for label, _, argument, expected in rows:
             self.assertLess(len(label), 120)
             self.assertEqual(end(argument), len(argument))
             self.assertEqual(end(expected), len(expected))
 
     def test_zero_divisor_and_shape_oracles_are_distinct(self):
         from numeric_batch import cases, error, n, pair, tree_list
         rows = cases()
         checked = next(expected for _, name, argument, expected in rows
                        if name == 'checkedBinary' and argument == pair(n(7), n(0)))
         self.assertTrue(checked.endswith(tree_list([error('non-zero')]*3)))
         self.assertNotEqual(error('non-zero'), n(0))
         self.assertNotEqual(error('not a natural number'), error('non-zero'))
 
 
 class BundleBatchTests(unittest.TestCase):
     def test_complete_independent_oracles(self):
         from bundle_batch import cases
         from source_order import end
         rows = cases()
         self.assertEqual(len(rows), len({row[0] for row in rows}))
         for label, _, _, arguments, expected in rows:
             self.assertLess(len(label), 120)
             for tree in arguments + [expected]:
                 self.assertEqual(end(tree), len(tree))
 
     def test_suffix_and_rejection_controls_remain_present(self):
         from bundle_batch import cases, byte_tree, n
         from source_order import field
         rows = cases()
         self.assertTrue(any(export == 'prefix' and field(expected, 1, 1) == byte_tree(b'TAIL')
                             for _, _, export, _, expected in rows))
         for code in [7, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18]:
             self.assertTrue(any(expected.startswith(b'\2\0\2') and field(expected, 1, 0) == n(code)
                                 for _, _, _, _, expected in rows), code)
 
 
 class ToolingBatchTests(unittest.TestCase):
     def test_complete_models_and_graph_ready_order(self):
         from tooling_batch_models import cases, graph_result, n, result
         from source_order import end
         # A newly ready earlier entry precedes the already-ready later one.
         self.assertEqual(graph_result([(0,[1]),(1,[]),(2,[])]), result(True,n([1,0,2])))
         self.assertEqual(graph_result([(0,[1]),(1,[2]),(2,[1])]), result(False,n(1)))
         rows=cases()
         self.assertEqual(len(rows),len({row[0] for row in rows}))
         for _,_,argument,expected in rows:
             for tree in [argument,expected]:self.assertEqual(end(tree),len(tree))
 
     def test_formatter_oracles_preserve_tokens_and_line_counts(self):
         from tooling_batch_models import format_cases
         import re
         tokens=re.compile(r'"(?:\\.|[^"\\])*"|\|-.*?-\||--[^\r\n]*|[^\s]',re.S)
         for label,source,expected in format_cases():
             self.assertEqual(tokens.findall(source),tokens.findall(expected),label)
             self.assertEqual(source.count('\n'),expected.count('\n'),label)
 
 
 class SourceFoundationOracleTests(unittest.TestCase):
     def test_corpus_has_complete_trees_and_unique_labels(self):
         from source_foundations_models import corpus
         from source_order import end
         rows = corpus(consumers=True)
         self.assertEqual(len(rows), len({row[0] for row in rows}))
         for label, _, argument, expected in rows:
             self.assertEqual(end(argument), len(argument), label)
             self.assertEqual(end(expected), len(expected), label)
 
     def test_shift_preserves_bound_indices_and_opaque_trees(self):
         from source_foundations_models import Node, shift, substitute
         bound = Node(4, Node(2, 0))
         free = Node(4, Node(2, 1))
         opaque = Node(5, free.tree())
         self.assertEqual(shift(bound, 7), bound)
         self.assertEqual(shift(free, 7), Node(4, Node(2, 8)))
         self.assertEqual(shift(opaque, 7), opaque)
         self.assertEqual(substitute(free, [Node(2, 3)], 5), Node(4, Node(2, 4)))
         self.assertEqual(shift(free, 2), shift(shift(free)))
 
 
 class CandidateResourceTests(unittest.TestCase):
     def test_memory_selection_reaches_heap_and_rss_guards(self):
         from collections_batch import ProbeRunner
         from unittest.mock import patch
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
             for selected in [8192, 65536]:
                 runner = ProbeRunner(root/'tricu', root, 60, memory_mib=selected)
                 with patch('source_order.run_guarded', return_value=('', 123)) as run:
                     runner.run('probe', ['run'], 30)
                     command = run.call_args.args[0]
                     self.assertIn(f'-M{selected}m', command)
                     self.assertEqual(run.call_args.kwargs['memory_mib'], selected)
                     self.assertLessEqual(run.call_args.kwargs['seconds'], 30)
             self.assertEqual(ProbeRunner(root/'tricu', root, 60).memory_mib, 8192)
 
     def test_memory_selection_rejects_out_of_policy_limits(self):
         from source_order import Runner
         for value in [-1, 0, 65537]:
             with self.assertRaises(ValueError):
                 Runner(Path('tricu'), Path('.'), memory_mib=value)
 
 
 class GrammarOracleTests(unittest.TestCase):
     def test_authored_replies_and_lookahead_vectors_are_whole_trees(self):
         from source_grammar_models import authored, lookahead_rows
         from source_order import end
         cases = authored()
         self.assertEqual(len(cases), len({row[0] for row in cases}))
         for label, _, argument, expected in cases:
             self.assertEqual(end(argument), len(argument), label)
             self.assertEqual(end(expected), len(expected), label)
         for row in lookahead_rows():
             for tree in row:
                 self.assertEqual(end(tree), len(tree))
 
     def test_layout_oracles_are_complete_and_independent(self):
         from source_layout_cases import semantic_cases, parser_cases
         from source_order import end
         values, replies = semantic_cases(), parser_cases()
         self.assertEqual(len(values + replies), len({row[0] for row in values + replies}))
         for label, _, value in values:
             self.assertEqual(end(value), len(value), label)
         for label, _, _, expected in replies:
             self.assertEqual(end(expected), len(expected), label)
         aligned = next(row[3] for row in replies if row[0] == 'ast-layout-2')
         deeper = next(row[3] for row in replies if row[0] == 'ast-layout-4')
         self.assertNotEqual(aligned, deeper)  # equal structure, different spans
         self.assertTrue(any(row[0] == 'unindented-body' for row in replies))
         self.assertTrue(any(row[0] == 'nested-do-indent' for row in replies))
 
     def test_lookahead_boundaries_do_not_consume_or_guess_assignments(self):
         from source_grammar_models import lookahead
         self.assertTrue(lookahead([2, 2, 5], [1, 3, 5])[0][0])
         self.assertFalse(lookahead([2, 6, 5], [1, 2, 3])[0][0])
         self.assertFalse(lookahead([2, 6, 11], [1, 2, 3])[0][1])
         self.assertTrue(lookahead([2, 3, 6, 4, 11], [1]*5)[0][1])
         self.assertFalse(lookahead([2, 12, 11], [1]*3)[0][1])
         self.assertEqual(lookahead([6, 2, 11], [1, 3, 5])[1],
                          [True, True, True, False, True, True])
 
 
 class ResolutionOracleTests(unittest.TestCase):
     def test_models_emit_complete_trees(self):
         from source_resolution_models import (ast_cases, scope_cases, pattern_cases,
             resolve_result, free, pattern_result, graph_cases, graph_model)
         from source_order import end
         for node, environment in ast_cases():
             for tree in [node.tree(), resolve_result(node, environment)]:
                 self.assertEqual(end(tree), len(tree))
         for node in scope_cases():
             for name in free(node):
                 self.assertEqual(name.kind, 2)
                 self.assertEqual(end(name.tree()), len(name.tree()))
         for node, binders, environment in pattern_cases():
             tree = pattern_result(node, binders, environment)
             self.assertEqual(end(tree), len(tree))
         for definitions, environment, final in graph_cases():
             for tree in graph_model(definitions, environment, final):
                 self.assertEqual(end(tree), len(tree))
 
     def test_scope_and_duplicate_priority(self):
         from source_resolution_models import Ast, free, resolved, graph_model
         from source_order import field, number
         name = Ast(2, 'x', 9)
         self.assertEqual(free(Ast(9, ('x', name, name))), [name])
         self.assertEqual(free(Ast(15, ('x', name, name))), [])
         bound = resolved(name, [], ('y', 'x', 'x'))
         self.assertEqual(number(field(bound, 1, 0)), 1)
         definitions = [(Ast(2, 'x', 1), Ast(2, 'missing', 2)),
                        (Ast(2, 'x', 3), Ast(1))]
         ordered, analyzed = graph_model(definitions, [], Ast(2, 'absent'))
         self.assertEqual(number(field(ordered, 1, 0, 0)), 30)
         self.assertEqual(number(field(analyzed, 1, 0, 0)), 20)
 
 
 class LoweringOracleTests(unittest.TestCase):
     def test_identity_and_constant(self):
         from source_lowering_models import tc_apply, I, K, n, LEAF
         for value in [LEAF, n(23), I]:
             self.assertEqual(tc_apply(I, value), value)
             self.assertEqual(tc_apply(tc_apply(K, value), n(99)), value)
 
     def test_complete_residual_models(self):
         from source_lowering_models import specimens, bracket, eliminate, run_result
         from source_order import end
         terms, cores = specimens()
         for tree in [bracket(t).tree() for t in terms] + [eliminate(c).tree() for c in cores] + [run_result(t) for t in terms]:
             self.assertEqual(end(tree), len(tree))
 
 
 class ModuleOracleTests(unittest.TestCase):
     def test_roundtrip_and_whole_expectations(self):
         from source_modules import tree_cases, decode, manifest_cases, header_cases
         from source_order import end
         from http_headers import result
         for tree in tree_cases():
             self.assertEqual(decode(list(tree), True), result(True, tree))
         for text, expected in manifest_cases() + header_cases():
             self.assertEqual(end(expected), len(expected))
 
 
 class BuildOracleTests(unittest.TestCase):
     def test_bundle_layout_and_missing_dependency(self):
         from source_building import bundle_bytes, map_cases
         from stable_bundle import sections
         from source_order import field
         from http_headers import LEAF
         with tempfile.TemporaryDirectory() as directory:
             path = Path(directory) / 'golden.arboricx'
             path.write_bytes(bundle_bytes([('x', LEAF), ('y', LEAF)]))
             self.assertEqual(len(sections(path)), 2)
         text, outcome = next((text, outcome) for text, outcome in map_cases() if '\tmissing\n' in text)
         self.assertEqual(field(outcome, 0), LEAF)
 
 
 class BackendDefaultTests(unittest.TestCase):
     def test_historical_provenance_is_not_reinterpreted(self):
         from generation_backend import DEFAULT_BACKEND, recorded_backend
         self.assertEqual(DEFAULT_BACKEND, 'stable')
         self.assertEqual(recorded_backend({}), 'optimized')
         for backend in ['stable', 'optimized']:
             self.assertEqual(recorded_backend({'backend': backend}), backend)
         with self.assertRaises(ValueError):
             recorded_backend({'backend': 'unknown'})
 
     def test_real_runner_parsers_default_stable_and_keep_overrides(self):
         # Stop at parse_args: exercise the actual CLI declarations without
         # reading artifacts, spawning hosts or building source in the fast gate.
         import argparse
         import runpy
         from unittest.mock import patch
 
         class Parsed(Exception):
             pass
 
         parse = argparse.ArgumentParser.parse_args
         scripts = {
             'bootstrap_generations.py': ['previous', 'output'],
             'bootstrap_generation_cli.py': ['previous'],
             'native_library_migration.py': ['tricu', 'generation', 'output', '--library', 'socket'],
             'portable_cli.py': ['tricu'],
             'source_equations.py': ['tricu', '--generation', 'generation'],
         }
         for script, required in scripts.items():
             for override in [None, 'stable', 'optimized']:
                 with self.subTest(script=script, override=override):
                     captured = []
                     argv = required + (['--backend', override] if override else [])
 
                     def intercept(parser, *args, **kwargs):
                         captured.append(parse(parser, argv))
                         raise Parsed()
 
                     with patch.object(argparse.ArgumentParser, 'parse_args', intercept):
                         with self.assertRaises(Parsed):
                             runpy.run_path(str(Path(__file__).parent / script), run_name='__main__')
                     self.assertEqual(captured[0].backend, override or 'stable')
 
 
 class HistoricalLibraryReferenceTests(unittest.TestCase):
     def test_historical_paths_are_labelled_without_source_substitution(self):
         import importlib.util
         spec = importlib.util.spec_from_file_location('site_reference', ROOT/'ext/demo-site/reference.py')
         module = importlib.util.module_from_spec(spec)
         spec.loader.exec_module(module)
         source, note = module.source_reference(ROOT, 'lib/io-v2.tri')
         self.assertIsNone(source)
         self.assertIn('Historical source path', note)
         source, note = module.source_reference(ROOT, 'lib/io.tri')
         self.assertEqual(source, ROOT/'lib/io.tri')
         self.assertIn('Source SHA-256', note)
         for path in ['lib/missing-current.tri', '../outside.tri', '/etc/passwd']:
             with self.assertRaises(ValueError):
                 module.source_reference(ROOT, path)
 
 
 class CheckpointProgressTests(unittest.TestCase):
     def test_private_and_unselected_files_are_never_read_as_checkpoints(self):
         from bootstrap_generations import published_checkpoints
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)
             cache = root / 'cas/cache/source-build'
             cache.mkdir(parents=True)
             (cache/'app').write_bytes(b'committed')
             # Directories would fail read_bytes if the watcher touched them.
             (cache/'473-17.tricu-stage-').mkdir()
             (cache/'unselected').mkdir()
             self.assertEqual(dict(published_checkpoints(root, ['app', 'not-yet-published'])),
                              {'app': b'committed'})
 
 
 class PublishedInventoryTests(unittest.TestCase):
     def test_only_reserved_unpublished_stage_names_are_excluded(self):
         from shared_library_generation import published_objects
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)
             shard = root / 'objects' / 'aaa'
             shard.mkdir(parents=True)
             (shard / ('a' * 64)).write_bytes(b'committed')
             (shard / '.tricu-stage-interrupted').write_bytes(b'partial')
             (shard / '473-17.tricu-stage-').write_bytes(b'GHC temporary naming')
             self.assertEqual(published_objects(root), {'a' * 64: b'committed'})
             (shard / ('a' * 63 + 'b')).write_bytes(b'extra committed object')
             self.assertEqual(len(published_objects(root)), 2)
             (shard / 'unexpected').write_bytes(b'not a reserved stage')
             with self.assertRaises(AssertionError):
                 published_objects(root)
 
 
 class RuntimeRetirementTests(unittest.TestCase):
     def test_generation_parser_rejects_retired_runtime_controls_before_output(self):
         import runpy
         import sys
         from contextlib import redirect_stderr
         from io import StringIO
         from unittest.mock import patch
         with tempfile.TemporaryDirectory() as temporary:
             output = Path(temporary) / 'not-created'
             for options in [('--execution-host', 'zig'), ('--zig-library', '/missing'),
                             ('--zig-gc-policy', '1', '1')]:
                 with self.subTest(options=options), patch.object(sys, 'argv',
                         ['bootstrap_generations.py', 'previous', str(output), *options]):
                     with redirect_stderr(StringIO()), self.assertRaises(SystemExit) as error:
                         runpy.run_path(str(Path(__file__).parent / 'bootstrap_generations.py'), run_name='__main__')
                     self.assertEqual(error.exception.code, 2)
                     self.assertFalse(output.exists())
 
 
 class BootstrapKitTests(unittest.TestCase):
     def setUp(self):
         from bootstrap_inputs import SCHEMA, ROLES
         self.temporary = tempfile.TemporaryDirectory()
         self.addCleanup(self.temporary.cleanup)
         self.directory = Path(self.temporary.name)
         self.metadata = dict(schema=SCHEMA, backend='stable', artifacts={})
         for role, filename in ROLES.items():
             payload = (role + '-opaque-test-bytes').encode()
             (self.directory / filename).write_bytes(payload)
             self.metadata['artifacts'][role] = dict(file=filename,
                 export='main' if role == 'adapter' else 'sourceMapBuildStable',
                 sha256=hashlib.sha256(payload).hexdigest())
         self.save()
 
     def save(self):
         (self.directory / 'bootstrap-kit.json').write_text(json.dumps(self.metadata))
 
     def load(self, backend='stable'):
         from bootstrap_inputs import load_kit
         return load_kit(self.directory, backend)
 
     def test_captures_exact_manifest_and_artifacts(self):
         raw, payloads, pin = self.load()
         self.assertEqual(raw, (self.directory / 'bootstrap-kit.json').read_bytes())
         self.assertEqual(pin, hashlib.sha256(raw).hexdigest())
         (self.directory / 'rebuilt-compiler.arboricx').write_bytes(b'changed')
         self.assertEqual(payloads['compiler'], b'compiler-opaque-test-bytes')
         with self.assertRaisesRegex(ValueError, 'digest mismatch'): self.load()
 
     def test_backend_is_explicit(self):
         with self.assertRaisesRegex(ValueError, 'backend mismatch'): self.load('optimized')
         self.metadata.pop('backend'); self.save()
         with self.assertRaisesRegex(ValueError, 'backend mismatch'): self.load()
 
     def test_rejects_invalid_manifest_schema(self):
         for schema in ('old', None, 1):
             with self.subTest(schema=schema):
                 self.metadata['schema'] = schema; self.save()
                 with self.assertRaisesRegex(ValueError, 'schema'): self.load()
 
     def test_requires_both_roles_and_no_extra_role(self):
         adapter = self.metadata['artifacts'].pop('adapter'); self.save()
         with self.assertRaisesRegex(ValueError, 'exactly'): self.load()
         self.metadata['artifacts']['adapter'] = adapter
         self.metadata['artifacts']['extra'] = adapter; self.save()
         with self.assertRaisesRegex(ValueError, 'exactly'): self.load()
 
     def test_rejects_path_traversal_and_role_swap(self):
         for filename in ('../elsewhere', '/tmp/elsewhere', 'io-adapter.arboricx'):
             with self.subTest(filename=filename):
                 self.metadata['artifacts']['compiler']['file'] = filename; self.save()
                 with self.assertRaisesRegex(ValueError, 'filename'): self.load()
 
     def test_rejects_wrong_export(self):
         self.metadata['artifacts']['compiler']['export'] = 'sourceMapBuild'; self.save()
         with self.assertRaisesRegex(ValueError, 'export'): self.load()
 
     def test_rejects_invalid_digest(self):
         for pin in ('a' * 63, 'A' * 64, '../x', None):
             with self.subTest(pin=pin):
                 self.metadata['artifacts']['adapter']['sha256'] = pin; self.save()
                 with self.assertRaisesRegex(ValueError, 'digest'): self.load()
 
     def test_explicit_versioned_kit(self):
         self.metadata.update(schema='tricu.bootstrap-kit.v2', io_version=2); self.save()
         self.load()
         for bad in [True, '2', 0, 3, None]:
             self.metadata['io_version'] = bad; self.save()
             with self.assertRaises(ValueError): self.load()
         self.metadata.pop('io_version'); self.save()
         with self.assertRaises(ValueError): self.load()
         self.metadata.update(schema='tricu.bootstrap-kit.v1', io_version=2); self.save()
         with self.assertRaises(ValueError): self.load()
 
     def test_missing_or_corrupt_adapter_fails(self):
         path = self.directory / 'io-adapter.arboricx'
         path.write_bytes(b'corrupt')
         with self.assertRaisesRegex(ValueError, 'digest mismatch'): self.load()
         path.unlink()
         with self.assertRaises(FileNotFoundError): self.load()
 
 
 class GenerationIOTests(unittest.TestCase):
     def test_strict_versions_and_historical_default(self):
         from generation_io import versions, version
         self.assertEqual(versions({}), (1, 1))
         for bad in [True, False, 0, 3, '2', None, 2.0]:
             with self.subTest(bad=bad), self.assertRaises(ValueError): version(bad)
         for incomplete in [{'execution_io_version': 2}, {'produced_adapter_sha256': 'a'*64}]:
             with self.assertRaises(ValueError): versions(incomplete)
         with self.assertRaises(ValueError): versions(dict(execution_io_version=2,
             produced_io_version=2, produced_adapter_sha256='A'*64))
 
     def test_host_and_authority_are_explicit(self):
         from generation_io import execution_flags, compiler_flags
         self.assertEqual(compiler_flags(1), [])
         self.assertEqual(compiler_flags(2), ['--toolchain-io-version', '2'])
         self.assertEqual(execution_flags(1), ['--unsafe-io'])
         with self.assertRaises(ValueError): execution_flags(1, 'haskell-ffi')
         self.assertEqual(execution_flags(2), [])
         self.assertEqual(execution_flags(2, 'haskell-ffi'), [])
         for host in ['browser', 'zig', 'unknown']:
             with self.assertRaises(ValueError): execution_flags(2, host)
 
     def test_invocation_grants_and_canonical_cli_selection(self):
         from generation_io import cli_version, check_invocation
         import argparse
         v2 = dict(execution_io_version=2, produced_io_version=2, produced_adapter_sha256='a'*64)
         check_invocation(v2, ['tricu', 'arboricx', 'run', '--io'])
         check_invocation({}, ['tricu', '--unsafe-io'])
         for flag in ['--unsafe-io', '--allow-read=/tmp', '--allow-write', '--allow-read-all', '--allow-write-all']:
             with self.assertRaises(ValueError): check_invocation(v2, ['tricu', flag])
         with self.assertRaises(ValueError): check_invocation({}, ['tricu', '--io'])
         for text in ['01', '+2', '2.0', ' 2', '3', '-1']:
             with self.assertRaises(argparse.ArgumentTypeError): cli_version(text)
         self.assertEqual(cli_version('2'), 2)
 
     def test_transition_chain_not_identical_execution_framing(self):
         from generation_io import check_pair
         first = dict(adapter_sha256='a'*64, execution_io_version=1,
                      produced_io_version=2, produced_adapter_sha256='b'*64)
         second = dict(adapter_sha256='b'*64, execution_io_version=2,
                       produced_io_version=2, produced_adapter_sha256='b'*64)
         check_pair(first, second)
         for key, value in [('execution_io_version', 1), ('produced_io_version', 1),
                            ('adapter_sha256', 'a'*64), ('produced_adapter_sha256', 'c'*64)]:
             with self.subTest(key=key), self.assertRaises(ValueError):
                 check_pair(first, {**second, key: value})
         check_pair(dict(adapter_sha256='a'*64), dict(adapter_sha256='a'*64))
 
     def test_produced_adapter_is_authenticated_not_input_adapter(self):
         from generation_io import produced_adapter
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
             (root/'io-adapter.arboricx').write_bytes(b'v1')
             (root/'produced-io-adapter.arboricx').write_bytes(b'v2')
             old = dict(adapter_sha256=hashlib.sha256(b'v1').hexdigest())
             new = dict(**old, execution_io_version=1, produced_io_version=2,
                        produced_adapter_sha256=hashlib.sha256(b'v2').hexdigest())
             self.assertEqual(produced_adapter(root, old), b'v1')
             self.assertEqual(produced_adapter(root, new), b'v2')
             (root/'produced-io-adapter.arboricx').write_bytes(b'v1')
             with self.assertRaises(ValueError): produced_adapter(root, new)
 
 
 class IndexedObjectAuditTests(unittest.TestCase):
     def test_independent_writer_and_complete_audit(self):
         from native_indexed_store import indexed_bytes
         from executable_object import verify_payload, INDEXED, FLAT
         for tree in [b'\0', b'\1\0', b'\2\1\0\2\0\0']:
             verify_payload(FLAT, tree)
             data = indexed_bytes(tree)
             verify_payload(INDEXED, data)
             for at in [0, 8, 15, 31, 32, 95, 96, len(data) - 1]:
                 changed = bytearray(data); changed[at] ^= 255
                 with self.subTest(tree=tree, at=at):
                     with self.assertRaises((AssertionError, ValueError)):
                         verify_payload(INDEXED, bytes(changed))
             for changed in [data[:-1], data + b'\0']:
                 with self.assertRaises(AssertionError): verify_payload(INDEXED, changed)
         with self.assertRaises(AssertionError): verify_payload('wrong.kind', b'\0')
 
 
 class InspectorStagingTests(unittest.TestCase):
     def test_staging_is_opt_in_immutable_and_authenticates_bytes(self):
         from library_inspector import stage_site
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)
             (root / 'producer').mkdir()
             site = root / 'site'; (site / 'bundles').mkdir(parents=True)
             index = site / 'packages.json'
             index.write_text(json.dumps(dict(schema=1, releases=[])))
             payload = root / 'producer/tricu-base.arboricx'
             payload.write_bytes(b'fixture bytes')
             digest = hashlib.sha256(payload.read_bytes()).hexdigest()
             report = dict(complete=True, releases={'tricu-base': dict(version='0.1.0', manifest='a'*64, bundle=digest)})
             with self.assertRaises(ValueError): stage_site(root, site, dict(report, complete=False))
             for version in (0, 3, True, "2"):
                 with self.assertRaises(ValueError): stage_site(root, site, dict(report, io_version=version))
             self.assertEqual(list((site / "bundles").iterdir()), [])
             stage_site(root, site, report)
             before = index.read_bytes(); stage_site(root, site, dict(report, io_version=2))
             self.assertEqual(index.read_bytes(), before)
             self.assertEqual((site / 'bundles' / (digest + '.arboricx')).read_bytes(), payload.read_bytes())
             payload.write_bytes(b'changed bytes')
             with self.assertRaises(ValueError): stage_site(root, site, report)
             changed = hashlib.sha256(payload.read_bytes()).hexdigest()
             report['releases']['tricu-base']['bundle'] = changed
             with self.assertRaises(ValueError): stage_site(root, site, report)
             self.assertEqual(index.read_bytes(), before)
             self.assertFalse((site / 'bundles' / (changed + '.arboricx')).exists())
             report['releases']['tricu-base']['version'] = '0.2.0'
             stage_site(root, site, report)
             self.assertEqual(len(json.loads(index.read_text())['releases']), 2)
             self.assertTrue((site / 'bundles' / (digest + '.arboricx')).exists())
 
 
 if __name__ == '__main__':
     unittest.main()