#!/usr/bin/env python3 """P13 reference evidence: read-only verification by default; collection is Root-only.""" import argparse import ast import inspect import json import os import re import stat import types from pathlib import Path import shutil import sys sys.dont_write_bytecode = True import run_all_tests as shared PROTOCOL = 'guarded-p13-reference-v1' CANONICAL_EVIDENCE = 'p13-reference-native-uid-corrected' NEW_TESTS = [ 'tests/special/test_peeping_room_geometry.gd', 'tests/special/test_peeping_room_reference.gd', 'tests/special/test_peeping_room_evidence.gd', 'tests/special/test_peeping_room_save.gd', 'tests/special/test_peeping_room_runtime.gd', ] ROSTER = NEW_TESTS + [ 'tests/s1/test_construction_service.gd', 'tests/s1/test_world_service.gd', 'tests/s1/test_path_service.gd', 'tests/s1/test_runtime_world_construction.gd', 'tests/supply/test_supply_save_migration.gd', 'tests/transport/test_transport_save.gd', 'tests/environment/test_runtime_weather.gd', 'tests/s6/test_runtime_snapshot_serialization.gd', 'tests/s6/test_runtime_save_lifecycle.gd', 'tests/supply/test_mimi_purchase_quote.gd', ] def uid_plan(refs): return {name + '.uid': {'owner': name, 'owner_sha256': pin['sha256'], 'file_type': 'regular'} for name, pin in sorted(refs.items()) if name.endswith('.gd') and name + '.uid' not in refs and (name.startswith('tests/special/') or name == 'scripts/special/peeping_room_service.gd')} def uid_snapshot(stage, source, plan): """Accept exactly the source inventory and finite native UID plan.""" actual = shared.inventory(stage, stage=True) shared.require(set(actual) == set(source) | set(plan), 'exact source and planned UID set') shared.require({n: actual[n] for n in source} == source, 'original source/settings unchanged') seen = set() for name in source: if name.endswith('.uid'): raw = (stage / name).read_bytes().rstrip(b'\r\n') shared.require(raw not in seen, 'existing project UID collision') seen.add(raw) result = {} for name, binding in plan.items(): path = stage / name shared.require(stat.S_ISREG(path.lstat().st_mode) and not path.is_symlink() and path.stat().st_nlink == 1, 'ordinary generated UID') raw = path.read_bytes() shared.require(re.fullmatch(rb'uid://[a-z0-9]{1,13}\n', raw) is not None, 'complete canonical UID bytes') shared.require(raw.rstrip(b'\n') not in seen, 'generated UID collision') seen.add(raw.rstrip(b'\n')) shared.require(stat.S_ISREG((stage / binding['owner']).lstat().st_mode) and shared.sha(stage / binding['owner']) == binding['owner_sha256'], 'UID owner content binding') result[name] = dict(binding, bytes=len(raw), raw_bytes=list(raw), sha256=shared.sha(path), text=raw.decode('ascii')) return result def install_uid_policy(namespace): """Specialize private globals; shared and P18 functions remain untouched.""" original_collect, original_seal = namespace['collect'], namespace['seal'] state = {} def reports(stage, expected, previous, requested): shared.require(not requested and not previous, 'P13 has no generated reports') snapshot = uid_snapshot(stage, state['source'], state['plan']) if state['accepted'] is None: shared.require(state['phase'] == 'import', 'UID generation only during import') state['accepted'] = snapshot expected.update({n: {'bytes': p['bytes'], 'sha256': p['sha256']} for n, p in snapshot.items()}) shared.require(snapshot == state['accepted'], 'sealed UID bytes unchanged') return shared.declared_reports(stage, expected, previous, requested) phase_globals = dict(vars(shared), declared_reports=reports) execute = types.FunctionType(shared.execute_phase.__code__, phase_globals) def phase(*args): stage, out, label = args[1:4] state['phase'] = label if label == 'import': shared.require(shared.inventory(stage, stage=True) == state['source'], 'original stage before import') before = {} shared.write(out / 'uid-plan.json', state['plan']) else: before = uid_snapshot(stage, state['source'], state['plan']) shared.require(before == state['accepted'], 'UID set before phase unchanged') row = execute(*args) after = uid_snapshot(stage, state['source'], state['plan']) shared.require(after == state['accepted'], 'UID set after phase unchanged') shared.write(out / (label + '-uids.json'), {'before': before, 'after': after}) return row namespace['shared'] = types.SimpleNamespace(**dict(vars(shared), execute_phase=phase)) def collect(args, root, out, refs, *rest): state.update(source=namespace['content_map'](refs), plan=uid_plan(refs), accepted=None) return original_collect(args, root, out, refs, *rest) def seal(out, refs, engine, contract_sha): shared.write(out / 'accepted-uids.json', state['accepted']) return original_seal(out, refs, engine, contract_sha) def retained(stage, source, out): plan = namespace['read_json'](out / 'uid-plan.json') shared.require(plan == uid_plan(source), 'independently derived exact UID plan') accepted = uid_snapshot(stage, source, plan) shared.require(namespace['read_json'](out / 'accepted-uids.json') == accepted, 'retained raw UID bytes and owner bindings') for label in ['import', 'parse'] + [str(i).zfill(3) for i in range(len(ROSTER))]: shared.require(namespace['read_json'](out / (label + '-uids.json')) == {'before': {} if label == 'import' else accepted, 'after': accepted}, 'all phase UID before/after seals') return True namespace.update(collect=collect, seal=seal, uid_retained=retained) def protocol_core(): """Reuse P18's complete guarded core without editing or copying its runner. P18 predates configurable protocol/roster constants. Specialize its three entry functions' identity literals and explicit roster cardinalities, plus one retained-stage comparison for the finite UID policy. The local phase adapter retains every original shared gate and seals its accepted sidecars. Fail closed if an upstream function changes any of these adapter sites. Both runner sources remain bound by the whole selected source map. """ core = shared.module('p13_p18_core', shared.HERE / 'run_p18_relationship_tests.py') namespace = dict(vars(core), ROSTER=ROSTER, __file__=__file__) replacements = { 'guarded-p18-v2': PROTOCOL, 'tools/run_p18_relationship_tests.py': 'tools/run_p13_reference_tests.py', 'exact seven unique tests': 'exact unique P13 roster', 'exact nine phases': 'exact P13 phase count', 'all nine captured engines': 'all P13 captured engines', } cardinalities = { 'len(set(ROSTER)) == 7': 'len(set(ROSTER)) == len(ROSTER)', 'len(phases) == 9': 'len(phases) == len(ROSTER) + 2', "len(tests) == 7": 'len(tests) == len(ROSTER)', } class Specialize(ast.NodeTransformer): def __init__(self): self.identities = 0 self.counts = 0 self.summary = 0 def visit_Constant(self, node): if isinstance(node.value, str) and node.value in replacements: self.identities += 1 return ast.copy_location(ast.Constant(replacements[node.value]), node) return node def visit_Compare(self, node): expression = ast.unparse(node) if expression in cardinalities: self.counts += 1 return ast.copy_location(ast.parse(cardinalities[expression], mode='eval').body, node) return self.generic_visit(node) def visit_Dict(self, node): for index, key in enumerate(node.keys): if isinstance(key, ast.Constant) and key.value == 'actual_suites': shared.require(isinstance(node.values[index], ast.Constant) and node.values[index].value == 7, 'upstream suite summary unchanged') node.values[index] = ast.parse('len(ROSTER)', mode='eval').body self.summary += 1 return self.generic_visit(node) expected = {'collect': (2, 0, 0), 'seal': (1, 0, 0), 'verify': (6, 3, 1)} for name, counts in expected.items(): adapter = Specialize() tree = adapter.visit(ast.parse(inspect.getsource(getattr(core, name)))) shared.require((adapter.identities, adapter.counts, adapter.summary) == counts, 'unchanged upstream adaptation sites: ' + name) if name == 'verify': old = "shared.inventory(confined(out, 'stage'), stage=True) == source" matches = [node for node in ast.walk(tree) if isinstance(node, ast.Compare) and ast.unparse(node) == old] shared.require(len(matches) == 1, 'unchanged retained-stage UID adapter site') class Retained(ast.NodeTransformer): def visit_Compare(self, node): if ast.unparse(node) == old: return ast.copy_location(ast.parse( "uid_retained(confined(out, 'stage'), source, out)", mode='eval').body, node) return self.generic_visit(node) tree = Retained().visit(tree) exec(compile(ast.fix_missing_locations(tree), str(shared.HERE / 'run_p18_relationship_tests.py'), 'exec'), namespace) install_uid_policy(namespace) return namespace def main(argv=None): argv = list(sys.argv[1:] if argv is None else argv) parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False) parser.add_argument('--output') modes = parser.add_mutually_exclusive_group() modes.add_argument('--collect', action='store_true', help='Root only; fresh native collection') modes.add_argument('--prepare-only', action='store_true') seen, index = set(), 0 while index < len(argv): flag = argv[index] shared.require(flag in ('--output', '--collect', '--prepare-only', '--help') and flag not in seen, 'unknown, duplicate or malformed flag') seen.add(flag) if flag == '--output': index += 1 shared.require(index < len(argv) and argv[index] and not argv[index].startswith('--'), 'output value required') index += 1 args = parser.parse_args(argv) shared.require(not args.collect or args.output is not None, 'Root collection requires explicit --output') if args.output is None: base = os.environ.get('BUILD_OUTPUT_DIR') shared.require(bool(base) and Path(base).is_absolute(), 'absolute BUILD_OUTPUT_DIR required') args.output = str(Path(base) / CANONICAL_EVIDENCE) args.timeout = 1200 # unchanged guarded core timeout; not a product value core = protocol_core() root = shared.HERE.parent out = core['plain_path'](args.output) refs = shared.source_refs(root) shared.require(len(set(ROSTER)) == len(ROSTER) and set(ROSTER).issubset(shared.roster(root)), 'complete unique discovered roster') markers = shared.module('p13_exact_markers', shared.HERE / 'strict_test_markers.py') contract_path = root / 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json' contract_sha = shared.sha(contract_path) contracts = markers.load_contracts(contract_path) shared.require(all(name in contracts for name in ROSTER), 'complete marker roster') policy = shared.read(shared.SUPPORT / 'suite-policy.json') shared.policy_coverage(ROSTER, refs, policy) for test in ROSTER: shared.phase_policy(test, Path(refs[test]['path']), policy['suite_policies']) for test in NEW_TESTS: shared.require(contracts[test]['source_sha256_at_schema_audit'] == refs[test]['sha256'], 'new strict source-hashed marker') parse_paths = ['res://' + name for name in sorted(refs) if name.endswith('.gd')] if args.prepare_only: ancestor = out.parent while not ancestor.exists(): ancestor = ancestor.parent free = shutil.disk_usage(ancestor).free size = sum(pin['bytes'] for pin in refs.values()) print(json.dumps(dict(protocol=PROTOCOL, roster=ROSTER, source_files=len(refs), source_bytes=size, free_bytes=free, conservative_required_bytes=size + shared.MIN_SPACE, capacity_available=free >= size + shared.MIN_SPACE, parse_paths=len(parse_paths), future_native_phases=len(ROSTER) + 2, actual_checks=None, engines=0, stages=0, formal_acceptance=False, original_unit_complete=False))) return 0 if args.collect: shared.require(not out.exists() and out.parent.is_dir(), 'fresh output with existing parent required') shared.require(not out.is_relative_to(root), 'private collection outside source tree') return core['collect'](args, root, out, refs, markers, contracts, policy, contract_path, contract_sha, parse_paths) return core['verify'](out, refs, markers, contracts, policy, contract_sha, parse_paths) if __name__ == '__main__': try: raise SystemExit(main()) except (AssertionError, OSError, ValueError, KeyError, TypeError) as error: print(type(error).__name__ + ': ' + str(error), file=sys.stderr) raise SystemExit(1)