284 lines
14 KiB
Python
284 lines
14 KiB
Python
#!/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)
|