#!/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 from pathlib import Path import shutil import sys sys.dont_write_bytecode = True import run_all_tests as shared PROTOCOL = 'guarded-p13-reference-v1' 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 protocol_core(): """Reuse P18's complete guarded core without editing or copying its runner. P18 predates configurable protocol/roster constants. Specialize ONLY its three entry functions' identity literals and explicit roster cardinalities. All proof gates and pure helpers are the imported original functions. 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) exec(compile(ast.fix_missing_locations(tree), str(shared.HERE / 'run_p18_relationship_tests.py'), 'exec'), 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) / 'p13-reference-native') 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)