diff --git a/config/extra-paths.json b/config/extra-paths.json index 6d5d07b..ba891be 100755 --- a/config/extra-paths.json +++ b/config/extra-paths.json @@ -1,4 +1,6 @@ { + "schema": "kam-source-resolution/1", + "adaptations": [], "source_paths": [], "library_paths": [] } diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 8c4821e..dd917d6 100755 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -89,10 +89,10 @@ class PipelineTests(unittest.TestCase): def test_partial_selection_never_means_whole_architecture_ready(self): p,r,_=self.run_pipeline(profiles='core_types') self.assertEqual(p.returncode,2,p.stdout);self.assertEqual(r['gate'],'PARTIAL_SELECTION') - def test_missing_sources_do_not_block_independent_platform_probe(self): + def test_missing_sources_block_the_entire_run(self): p,r,_=self.run_pipeline(profiles='core_types',with_source=False) self.assertEqual(p.returncode,2,p.stdout);self.assertEqual(r['gate'],'BLOCKED') - self.assertEqual(r['profiles'][0]['compile_link']['status'],'PASS') + self.assertEqual(r['profiles'][0]['compile_link']['status'],'BLOCKED') self.assertEqual(r['profiles'][1]['compile_link']['status'],'BLOCKED') def test_no_resource_data_in_evidence(self): import zipfile diff --git a/tests/test_probe.py b/tests/test_probe.py index b98ad5c..a724f66 100755 --- a/tests/test_probe.py +++ b/tests/test_probe.py @@ -1,9 +1,10 @@ from __future__ import annotations -import hashlib, json, os, shutil, subprocess, sys, tempfile, unittest, zipfile +import hashlib, json, os, shutil, subprocess, sys, tempfile, unittest, xml.etree.ElementTree as ET, zipfile from pathlib import Path from unittest.mock import patch sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools')) from common import ROOT, COMMIT, ProbeError, git_blob, safe_relative, sha256, write_json, run_command, classify +from source_resolution import build_source_resolution, load_source_contract, validate_source_contract from source_fetch import selected, enumerate_blobs, verify from source_audit import inventory from generate import make_target, imports_source, asset_inventory, COMPILER_OPTIONS @@ -22,6 +23,62 @@ for i,good in enumerate(('src/common/KM_Points.pas','Maps/Тест/Тест.map' def test(self,good=good):self.assertEqual(safe_relative(good),good) setattr(PathsTests,f'test_accept_path_{i:02d}',test) +class SourceContractTests(unittest.TestCase): + def contract(self, value): + temp = tempfile.TemporaryDirectory(); self.addCleanup(temp.cleanup) + path = Path(temp.name) / 'extra-paths.json'; path.write_text(json.dumps(value)) + return path + def test_loads_normalized_targets_and_deduplicated_paths(self): + path = self.contract({'schema':'kam-source-resolution/1','adaptations':['src/common/KM_Example.pas'], + 'source_paths':['tools','tools'],'library_paths':[str(ROOT/'config')]}) + loaded = load_source_contract(path) + self.assertEqual(loaded['adaptation_targets'], ('src/common/KM_Example.pas',)) + self.assertEqual(loaded['source_paths'], ((ROOT/'tools').resolve(),)) + self.assertEqual(loaded['library_paths'], ((ROOT/'config').resolve(),)) + def test_rejects_legacy_and_unsafe_contracts(self): + for value in ({'source_paths':[],'library_paths':[]}, + {'schema':'kam-source-resolution/1','adaptations':['src\\bad.pas'],'source_paths':[],'library_paths':[]}): + with self.subTest(value=value), self.assertRaises(ProbeError): load_source_contract(self.contract(value)) + +class SourceOwnershipTests(unittest.TestCase): + def setUp(self): + self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name);self.vendor=self.root/'vendor';self.sparse=self.root/'port-source/kam_remake' + self.vendor.joinpath('src').mkdir(parents=True);self.vendor.joinpath('src/KM_Test.pas').write_text('unit KM_Test; interface implementation end.') + self.receipt={'files':[{'path':'src/KM_Test.pas'}]};self.inventory={'units':{'km_test':['src/KM_Test.pas']}} + def tearDown(self):self.tmp.cleanup() + def validate(self, contract): + with patch('source_resolution.ROOT',self.root): + return validate_source_contract(contract,self.vendor,self.receipt,self.inventory) + def test_registered_adaptation_has_vendor_identity_and_unit(self): + self.sparse.joinpath('src').mkdir(parents=True);self.sparse.joinpath('src/KM_Test.pas').write_text('unit km_test; interface implementation end.') + result=self.validate({'adaptation_targets':('src/KM_Test.pas',),'source_paths':(),'library_paths':()}) + self.assertEqual(result['adaptations'][0]['unit_name'],'km_test');self.assertEqual(result['vendor_unit_keys'],frozenset({'km_test'})) + def test_rejects_unregistered_sparse_and_generic_shadowing(self): + self.sparse.mkdir(parents=True);self.sparse.joinpath('other.inc').write_text('x') + with self.assertRaises(ProbeError):self.validate({'adaptation_targets':(),'source_paths':(),'library_paths':()}) + shutil.rmtree(self.sparse);generic=self.root/'generic';generic.mkdir();generic.joinpath('KM_Test.pas').write_text('not a unit') + with self.assertRaises(ProbeError):self.validate({'adaptation_targets':(),'source_paths':(generic,),'library_paths':()}) + +class CompilerSourceResolutionTests(unittest.TestCase): + def test_orders_deduplicates_and_guards_application_vendor_units(self): + with tempfile.TemporaryDirectory() as t: + root = Path(t); app = root/'app'; vendor = root/'vendor'; generic = root/'generic'; library = root/'library' + for path in (app, vendor/'src', vendor/'ext', generic, library): path.mkdir(parents=True, exist_ok=True) + (vendor/'src/KM_Vendor.pas').write_text('unit KM_Vendor; interface implementation end.') + first, second = root/'adaptations/z', root/'adaptations/a' + first.mkdir(parents=True); second.mkdir(parents=True) + validated = {'vendor_unit_keys': frozenset({'km_vendor'}), 'adaptations': ( + {'target':'z.pas', 'project_file':first/'z.pas'}, {'target':'A.pas', 'project_file':second/'A.pas'}, + {'target':'a2.pas', 'project_file':second/'a2.pas'}), + 'source_paths': (generic, second), 'library_paths': (library,)} + result = build_source_resolution(validated, vendor, app) + self.assertEqual(result['adaptation_dirs'], (second.resolve(), first.resolve())) + self.assertEqual(result['compiler_source_paths'][:4], + (app.resolve(), second.resolve(), first.resolve(), generic.resolve())) + self.assertEqual(result['library_paths'], (library.resolve(),)) + (app/'KM_VENDOR.pas').write_text('malformed declaration') + with self.assertRaises(ProbeError): build_source_resolution(validated, vendor, app) + class SourceSelectionTests(unittest.TestCase): def test_known_git_blob(self): self.assertEqual(git_blob(b'hello\n'),'ce013625030ba8dba906f756967f9e9ca394464a') @@ -177,10 +234,25 @@ class GenerationTests(unittest.TestCase): code,units=imports_source(PROFILES[1]);self.assertIn('KMPointRight(P)',code);self.assertIn('KM_Points',units) def test_actual_stream_calls(self): code,units=imports_source(PROFILES[2]);self.assertIn('TKMemoryStreamBinary.Create',code);self.assertIn('S.ReadW(U)',code) + def test_upstream_manifest_uses_resolved_compiler_order(self): + with tempfile.TemporaryDirectory() as t: + root=Path(t);vendor=root/'vendor';generic=root/'generic';library=root/'library';adaptation=root/'adaptation' + for path in (vendor/'src',generic,library,adaptation):path.mkdir(parents=True,exist_ok=True) + (vendor/'src/KM_Points.pas').write_text('unit KM_Points; interface implementation end.') + validated={'vendor_unit_keys':frozenset({'km_points'}),'adaptations':( + {'target':'src/KM_Points.pas','project_file':adaptation/'KM_Points.pas'},), + 'source_paths':(generic,),'library_paths':(library,)} + p=root/'target';rec=make_target(p,PROFILES[1],'aarch64','id',vendor,validated) + self.assertEqual(rec['source_paths'][:3],[str((p/'code').resolve()),str(adaptation.resolve()),str(generic.resolve())]) + xml=ET.parse(p/'CastleEngineManifest.xml') + self.assertEqual([node.attrib['value'] for node in xml.findall('./compiler_options/search_paths/path')][:3], + ['code',os.path.relpath(adaptation,p).replace(os.sep,'/'),os.path.relpath(generic,p).replace(os.sep,'/')]) + self.assertEqual([node.attrib['value'] for node in xml.findall('./compiler_options/library_paths/path')], + [os.path.relpath(library,p).replace(os.sep,'/')]) def test_manifest_no_standalone_target(self): with tempfile.TemporaryDirectory() as t: p=Path(t)/'target' - rec=make_target(p,PROFILES[0],'aarch64','test-build',None,{'source_paths':[],'library_paths':[]}) + rec=make_target(p,PROFILES[0],'aarch64','test-build',None,None) xml=(p/'CastleEngineManifest.xml').read_text() self.assertNotIn('standalone_source',xml);self.assertNotIn('test_shims',xml) self.assertIn('game_units="GameInitialize"',xml) @@ -190,7 +262,7 @@ class GenerationTests(unittest.TestCase): def test_decision_profile_uses_existing_generator_with_alternate_game_unit(self): with tempfile.TemporaryDirectory() as t: profile=next(x for x in PROFILES if x['id']=='oq_host_callback') - p=Path(t)/'target';rec=make_target(p,profile,'aarch64','id',None,{'source_paths':[],'library_paths':[]}) + p=Path(t)/'target';rec=make_target(p,profile,'aarch64','id',None,None) xml=(p/'CastleEngineManifest.xml').read_text() self.assertIn('game_units="DecisionHostCallback"',xml) self.assertEqual(rec['game_unit'],'DecisionHostCallback') @@ -204,13 +276,13 @@ class GenerationTests(unittest.TestCase): self.assertEqual([x['path'] for x in rows],['Maps/one.txt']) def test_fresh_target_required(self): with tempfile.TemporaryDirectory() as t: - with self.assertRaises(ProbeError):make_target(Path(t),PROFILES[0],'aarch64','id',None,{'source_paths':[],'library_paths':[]}) + with self.assertRaises(ProbeError):make_target(Path(t),PROFILES[0],'aarch64','id',None,None) def test_optional_assets_preserved(self): with tempfile.TemporaryDirectory() as t: r=Path(t)/'assets';r.mkdir();(r/'Data').mkdir();raw=bytes(range(256)) (r/'Data/sample.dat').write_bytes(raw) entries=asset_inventory(r) - p=Path(t)/'target';make_target(p,PROFILES[0],'aarch64','id',None,{'source_paths':[],'library_paths':[]},r,entries) + p=Path(t)/'target';make_target(p,PROFILES[0],'aarch64','id',None,None,r,entries) self.assertEqual((p/'data/game/Data/sample.dat').read_bytes(),raw) def test_asset_symlinks_rejected(self): with tempfile.TemporaryDirectory() as t: diff --git a/tools/generate.py b/tools/generate.py index c71c973..710de2b 100755 --- a/tools/generate.py +++ b/tools/generate.py @@ -3,7 +3,7 @@ from __future__ import annotations import json, os, shutil, urllib.parse, xml.etree.ElementTree as ET from pathlib import Path from common import ROOT, COMMIT, VERSION, CPUS, ProbeError, safe_relative, sha256, write_json -from source_audit import search_paths +from source_resolution import build_source_resolution UPSTREAM_DEFINES=['DBG_SKIP_SECURE_AUTH'] # Existing upstream switch; private multiplayer authentication is out of scope. COMPILER_OPTIONS=['-B','-CX-','-XX-','-k--no-undefined','-k--no-gc-sections'] @@ -76,7 +76,7 @@ def asset_inventory(root:Path|None, includes:list[str]|None=None)->list[dict]: return result def make_target(project:Path,profile:dict,cpu:str,build_id:str,source:Path|None, - extras:dict,assets_root:Path|None=None,assets:list[dict]|None=None)->dict: + validated:dict|None,assets_root:Path|None=None,assets:list[dict]|None=None)->dict: if project.exists():raise ProbeError('Target directory must be fresh: '+str(project)) project.mkdir(parents=True) shutil.copytree(ROOT/'code',project/'code') @@ -100,13 +100,15 @@ def make_target(project:Path,profile:dict,cpu:str,build_id:str,source:Path|None, game_units=profile.get('game_unit','GameInitialize'),compiler='fpc',screen_orientation='landscape',fullscreen_immersive='true') ET.SubElement(root,'version',value=VERSION,code='30') options=ET.SubElement(root,'compiler_options');paths=ET.SubElement(options,'search_paths') - ET.SubElement(paths,'path',value='code/') - include=[] if units: - if source is None:raise ProbeError('KaM source is required for '+profile['id']) - include=search_paths(source) - include.extend(Path(p).resolve() for p in extras.get('source_paths',[])) - for p in include: + if source is None or validated is None:raise ProbeError('KaM source resolution is required for '+profile['id']) + resolution=build_source_resolution(validated,source,project/'code') + compiler_source_paths=resolution['compiler_source_paths'] + library_paths=resolution['library_paths'] + else: + compiler_source_paths=(project/'code',) + library_paths=() + for p in compiler_source_paths: if not p.is_dir():raise ProbeError('Source path missing: '+str(p)) ET.SubElement(paths,'path',value=os.path.relpath(p,project).replace(os.sep,'/')) defines=ET.SubElement(options,'defines') @@ -114,15 +116,14 @@ def make_target(project:Path,profile:dict,cpu:str,build_id:str,source:Path|None, custom=ET.SubElement(options,'custom_options') for flag in COMPILER_OPTIONS:ET.SubElement(custom,'option').text=flag libraries=ET.SubElement(options,'library_paths') - for p in extras.get('library_paths',[]): - p=Path(p).resolve() + for p in library_paths: if not p.is_dir():raise ProbeError('Native library path missing: '+str(p)) ET.SubElement(libraries,'path',value=os.path.relpath(p,project).replace(os.sep,'/')) # Android SDK versions deliberately inherited from the locked CGE image. ET.indent(root) ET.ElementTree(root).write(project/'CastleEngineManifest.xml',encoding='utf-8',xml_declaration=True) description={'profile':profile['id'],'name':name,'package':'org.kammobile.archprobe.'+profile['id'].replace('_',''), - 'cpu':cpu,'units':units,'game_unit':profile.get('game_unit','GameInitialize'),'source_paths':[str(p) for p in include], + 'cpu':cpu,'units':units,'game_unit':profile.get('game_unit','GameInitialize'),'source_paths':[str(p) for p in compiler_source_paths], 'compiler_options':COMPILER_OPTIONS,'upstream_defines':UPSTREAM_DEFINES if units else [], 'assets':entries,'shims':False} write_json(project/'target.json',description) diff --git a/tools/runner.py b/tools/runner.py index a387fbd..63546d3 100755 --- a/tools/runner.py +++ b/tools/runner.py @@ -5,6 +5,7 @@ from pathlib import Path from common import ROOT, COMMIT, CPUS, ProbeError, stamp, sha256, write_json, run_command, first_errors, classify from source_fetch import fetch, verify from source_audit import inventory +from source_resolution import load_source_contract, validate_source_contract from generate import make_target, asset_inventory from artifacts import inspect_elf, inspect_apk @@ -93,14 +94,6 @@ def new_result(cpu:str,profiles:list[dict],reason:str='Not started')->tuple[dict 'desktop_target':False,'shims':False,'runtime':'NOT_RUN', 'profiles':[blocked_row(p,reason) for p in profiles]},out) -def extra_paths()->dict: - d=json.loads((ROOT/'config/extra-paths.json').read_text()) - for key in ('source_paths','library_paths'): - if not isinstance(d.get(key),list) or not all(isinstance(s,str) for s in d[key]): - raise ProbeError('extra-paths.json must contain string arrays') - d[key]=[str((ROOT/p).resolve()) if not Path(p).is_absolute() else str(Path(p).resolve()) for p in d[key]] - return d - def tool_versions(out:Path)->dict: commands={'castle':['castle-engine','--version'],'fpc':['fpc','-iV'],'java':['java','-version'], 'adb':['adb','version']} @@ -135,10 +128,10 @@ def build_local(a:argparse.Namespace,profiles:list[dict])->int: if not shutil.which('castle-engine'): for r in result['profiles']:r['compile_link']['reason']='castle-engine not installed; no Android compilation attempted' return finalize(result,out) - source=ROOT/'vendor/kam_remake';audit=None;source_error='';assets=[] + source=ROOT/'vendor/kam_remake';audit=None;validated=None;assets=[] need_source=any(p['units'] for p in profiles) try: - extras=extra_paths() + contract=load_source_contract() assets=asset_inventory(Path(a.assets),a.asset_path) if a.assets else [] result['asset_count']=len(assets) if need_source: @@ -147,13 +140,10 @@ def build_local(a:argparse.Namespace,profiles:list[dict])->int: else:receipt=fetch(source) write_json(out/'source-receipt.json',receipt) audit=inventory(source);write_json(out/'source-inventory.json',audit) - for path in extras['source_paths']: - for p in Path(path).rglob('*.pas'): - if p.stem.casefold() in audit['units']: - raise ProbeError('Extra path shadows original unit '+p.stem+'; no shims allowed') + validated=validate_source_contract(contract,source,receipt,audit) except (ProbeError,OSError,ValueError,KeyError) as e: - source_error=str(e);extras={'source_paths':[],'library_paths':[]} - result['source_or_inputs_error']=source_error + result['source_or_inputs_error']=str(e) + return finalize(result,out) work=ROOT/'build'/result['run_id'];work.mkdir(parents=True) dist=ROOT/'dist'/result['run_id'];dist.mkdir(parents=True) package_index=[] @@ -162,14 +152,13 @@ def build_local(a:argparse.Namespace,profiles:list[dict])->int: print(f"[{idx+1}/{len(profiles)}] Android {a.cpu}: {ident}",flush=True) try: if profile['units']: - if source_error:raise ProbeError(source_error) missing=[u for u in profile['units'] if u.casefold() not in audit['units']] if missing:raise ProbeError('Missing required original source units: '+', '.join(missing)) # Resources only enter APKs when explicitly requested. The platform baseline remains independent. pack=profile['package'] or a.package_all use_assets=bool(a.assets) and pack and ident=='platform' description=make_target(project,profile,a.cpu,result['run_id'],source if profile['units'] else None, - extras if profile['units'] else {'source_paths':[],'library_paths':[]}, + validated if profile['units'] else None, Path(a.assets) if use_assets else None,assets if use_assets else []) row['target']=description evidence=out/ident;evidence.mkdir() @@ -225,7 +214,7 @@ def build_local(a:argparse.Namespace,profiles:list[dict])->int: else:row['compile_link']={'status':'BLOCKED','reason':str(e)} finally: write_json(out/'result-in-progress.json',result) - if need_source and not source_error: + if need_source: try:verify(source) except ProbeError as e: result['source_modified_during_build']=str(e) diff --git a/tools/source_resolution.py b/tools/source_resolution.py new file mode 100644 index 0000000..486b1dd --- /dev/null +++ b/tools/source_resolution.py @@ -0,0 +1,189 @@ +"""Load the permanent source-resolution contract.""" +from __future__ import annotations + +import json +import re +from pathlib import Path + +from common import ROOT, ProbeError, safe_relative +import source_audit + + +_KEYS = {'schema', 'adaptations', 'source_paths', 'library_paths'} +_SCHEMA = 'kam-source-resolution/1' +_SOURCE_SUFFIXES = {'.pas', '.pp', '.inc', '.inl'} +_UNIT_DECLARATION = re.compile(r'^\s*unit\s+([\w.]+)\s*;', re.I | re.M) + + +def _unit_name(path: Path) -> str | None: + match = _UNIT_DECLARATION.search(path.read_text(encoding='utf-8-sig', errors='replace')) + return match[1] if match else None + + +def _inside(path: Path, root: Path) -> bool: + return path == root or path.is_relative_to(root) + + +def validate_source_contract( + contract: dict, + vendor_root: Path, + vendor_receipt: dict, + vendor_inventory: dict, +) -> dict: + """Validate sparse overrides and generic source paths before compilation.""" + sparse_root = ROOT / 'port-source/kam_remake' + if sparse_root.exists() and sparse_root.is_symlink(): + raise ProbeError(f'Sparse source root is linked: {sparse_root}') + try: + sparse_resolved = sparse_root.resolve() + vendor_resolved = vendor_root.resolve() + except (OSError, ValueError, RuntimeError) as error: + raise ProbeError(f'Cannot resolve source roots: {error}') from error + + receipt_paths = {entry.get('path') for entry in vendor_receipt.get('files', []) + if isinstance(entry, dict) and isinstance(entry.get('path'), str)} + targets = contract.get('adaptation_targets', ()) + registered = set() + adaptations = [] + for target in targets: + if not isinstance(target, str) or target.casefold() in registered: + raise ProbeError(f'Duplicate or invalid adaptation target: {target!r}') + registered.add(target.casefold()) + try: + target = safe_relative(target) + except ProbeError: + raise + if Path(target).suffix.casefold() not in _SOURCE_SUFFIXES: + raise ProbeError(f'Unsupported adaptation target: {target}') + project_file, vendor_file = sparse_root / target, vendor_root / target + if target not in receipt_paths: + raise ProbeError(f'Adaptation is not an exact vendor receipt path: {target}') + if not project_file.is_file() or project_file.is_symlink(): + raise ProbeError(f'Registered adaptation missing or linked: {project_file}') + for parent in (project_file.parent, *project_file.parents): + if parent == sparse_root.parent: + break + if parent.exists() and parent.is_symlink(): + raise ProbeError(f'Linked component in adaptation path: {parent}') + try: + if not _inside(project_file.resolve(), sparse_resolved): + raise ProbeError(f'Adaptation escapes sparse root: {project_file}') + except (OSError, ValueError, RuntimeError) as error: + raise ProbeError(f'Cannot resolve adaptation: {project_file}') from error + if not vendor_file.is_file(): + raise ProbeError(f'Vendor adaptation missing: {vendor_file}') + unit_name = None + if project_file.suffix.casefold() in {'.pas', '.pp'}: + unit_name, vendor_unit = _unit_name(project_file), _unit_name(vendor_file) + if not unit_name or not vendor_unit or unit_name.casefold() != vendor_unit.casefold(): + raise ProbeError(f'Adaptation unit mismatch: {target}') + if len(vendor_inventory.get('units', {}).get(unit_name.casefold(), [])) > 1: + raise ProbeError(f'Ambiguous vendor unit adaptation: {unit_name}') + adaptations.append({'target': target, 'project_file': project_file, + 'vendor_file': vendor_file, 'unit_name': unit_name}) + + if sparse_root.exists(): + for path in sparse_root.rglob('*'): + if path.suffix.casefold() not in _SOURCE_SUFFIXES: + continue + if path.is_symlink() or not path.is_file(): + raise ProbeError(f'Linked or non-regular sparse source: {path}') + if path.relative_to(sparse_root).as_posix() not in {item['target'] for item in adaptations}: + raise ProbeError(f'Unregistered sparse source: {path}') + + vendor_units = vendor_inventory.get('units', {}) + source_paths = tuple(contract.get('source_paths', ())) + for source_path in source_paths: + if not source_path.is_dir(): + raise ProbeError(f'Generic source path is not a directory: {source_path}') + resolved = source_path.resolve() + if _inside(resolved, vendor_resolved) or _inside(resolved, sparse_resolved): + raise ProbeError(f'Generic source path is protected: {source_path}') + found = source_audit.inventory(source_path) + if set(found['units']) & set(vendor_units): + raise ProbeError(f'Generic source path shadows vendor unit: {source_path}') + for path in source_path.rglob('*'): + if path.is_file() and path.suffix.casefold() in {'.pas', '.pp'} and path.stem.casefold() in vendor_units: + raise ProbeError(f'Generic source filename shadows vendor unit: {path}') + + library_paths = tuple(contract.get('library_paths', ())) + if any(not path.is_dir() for path in library_paths): + raise ProbeError('Library path is not a directory') + return {'adaptations': tuple(adaptations), 'vendor_unit_keys': frozenset(vendor_units), + 'source_paths': source_paths, 'library_paths': library_paths} + + +def build_source_resolution( + validated: dict, + vendor_root: Path, + application_source: Path, +) -> dict: + """Project validated source inputs into deterministic compiler precedence.""" + try: + vendor_root, application_source = vendor_root.resolve(), application_source.resolve() + if not vendor_root.is_dir() or not application_source.is_dir(): + raise ProbeError('Vendor root and application source must exist as directories') + + vendor_unit_keys = validated['vendor_unit_keys'] + if set(source_audit.inventory(application_source)['units']) & vendor_unit_keys: + raise ProbeError('Application source cannot replace vendor units; put replacements under ' + 'port-source/kam_remake/ and register them in adaptations') + for path in application_source.rglob('*'): + if path.is_file() and path.suffix.casefold() in {'.pas', '.pp'} and path.stem.casefold() in vendor_unit_keys: + raise ProbeError('Application source cannot replace vendor units; put replacements under ' + 'port-source/kam_remake/ and register them in adaptations') + + adaptation_dirs = tuple(dict.fromkeys( + record['project_file'].parent.resolve() + for record in sorted(validated['adaptations'], key=lambda record: (record['target'].casefold(), record['target'])) + )) + generic_source_paths = tuple(path.resolve() for path in validated['source_paths']) + vendor_fallback_paths = tuple(path.resolve() for path in source_audit.search_paths(vendor_root)) + compiler_source_paths = tuple(dict.fromkeys((application_source, *adaptation_dirs, + *generic_source_paths, *vendor_fallback_paths))) + library_paths = tuple(path.resolve() for path in validated['library_paths']) + except ProbeError: + raise + except (AttributeError, KeyError, OSError, RuntimeError, TypeError, ValueError) as error: + raise ProbeError(f'Cannot construct compiler source order: {error}') from error + return {'application_source': application_source, 'adaptation_dirs': adaptation_dirs, + 'generic_source_paths': generic_source_paths, 'vendor_fallback_paths': vendor_fallback_paths, + 'compiler_source_paths': compiler_source_paths, 'library_paths': library_paths} + + +def load_source_contract(config_path: Path | None = None) -> dict: + """Return validated, immutable source-resolution inputs.""" + path = config_path or ROOT / 'config/extra-paths.json' + try: + value = json.loads(path.read_text(encoding='utf-8')) + except (OSError, UnicodeError, json.JSONDecodeError) as error: + raise ProbeError(f'Cannot read source-resolution contract: {error}') from error + if not isinstance(value, dict) or set(value) != _KEYS: + raise ProbeError('Invalid source-resolution contract fields') + if value['schema'] != _SCHEMA: + raise ProbeError('Invalid source-resolution schema') + for key in _KEYS - {'schema'}: + if not isinstance(value[key], list) or not all(isinstance(item, str) for item in value[key]): + raise ProbeError(f'{key} must be an array of strings') + + try: + adaptations = tuple(safe_relative(target) for target in value['adaptations']) + except ProbeError: + raise + + def paths(key: str) -> tuple[Path, ...]: + result = [] + for item in value[key]: + if not item: + raise ProbeError(f'{key} cannot contain an empty path') + try: + candidate = Path(item) + resolved = (candidate if candidate.is_absolute() else ROOT / candidate).resolve() + except (OSError, ValueError, RuntimeError) as error: + raise ProbeError(f'Invalid {key} path: {item!r}') from error + if resolved not in result: + result.append(resolved) + return tuple(result) + + return {'schema': _SCHEMA, 'adaptation_targets': adaptations, + 'source_paths': paths('source_paths'), 'library_paths': paths('library_paths')}