#!/usr/bin/env python3 """Read-only P04 evidence preflight. Never grants original acceptance. Required conditions and authority bindings are fixed here, never supplied by an index. See execution/p04-01/CHECKER_PROTOCOL.md for the supported input subset. """ import argparse from contextlib import contextmanager import hashlib import json import os from pathlib import Path import re import subprocess import sys import tempfile sys.dont_write_bytecode = True ROOT = Path(__file__).resolve().parents[1] SPEC = ROOT / 'docs/spec/acceptance/P04-01.json' SPEC_HASH = '9ec722e7dfb0bee178d291624a026420c08412574e238bba1fde012c7a6f7881' DOMAINS = {'閉店', '停電', '破壊', '退出', '課金', 'シフト交代', '00:00決算'} REGISTERED_COMMIT = 'deb1bcb2374eb378b840ca345db952bd8572cb6e' EXECUTION_ORIGIN = '9767b0a28ca8d252520307dbbac60a1d8cfb87ab' CHECKER_PATH = 'tools/check_p04_order_evidence.py' DOC_SCOPE = 'docs/plans/original-completion-2026-10-02/execution/p04-01/' OWNER_BINDINGS = { ref: (DOC_SCOPE + leaf, digest) for refs, leaf, digest in ( (('D05-P04-DEPENDENCY', 'D05-P04-SHIFT-1800'), 'D05_OWNER_APPROVAL.json', '4583ee9aad45c4db183d72df4420df48aa8e0897e3b5efda13016f0b5cdbd68e'), (('D05-P04-DESTRUCTION', 'D05-P04-MIDNIGHT-DAY', 'D05-P04-PAUSE-REASON', 'D05-P04-EXIT'), 'D05_REMAINING_APPROVAL.json', '09b5092124574c3ccbf3b55169729a7529d9dde9b9a115d5e944fce4b960f5e4'), ) for ref in refs } MAIN_ANCHOR = 'c0ea009b28a7649aca6a433a1a24dc0be5e7ab81' # Documentation only: external Runner prepare_integration/integrate and Root # normal_c_join joins after final. The checker never consumes this roster. NORMAL_C_PROVENANCE_JOIN = frozenset({ 'candidate_commit', 'candidate_tree', 'first_parent', 'second_parent', 'tested_tree', 'prepare_integration_receipt', 'independent_review_receipt', 'validation_receipt', 'source_manifest_sha256', 'settings_sha256', 'markers_sha256', 'policy_sha256', 'engine_sha256', 'inputs_sha256', }) SELECTION_SHA = '320ca668732b4ff72fc4449998af2327d4e5c527b06896740806aaf0487282b4' _GIT_ENV = None # Fixed membership, never supplied by the evidence index. ORDER_FUNCTIONS = { 'scripts/core/simulation_kernel.gd': ('advance_one_minute', '_run_phase', '_handle_time_boundaries'), 'scripts/core/command_queue.gd': ('_comes_before',), 'autoload/simulation_runtime.gd': ( '_process_tick_commands', '_advance_operations_time', '_run_operations_deferred_suffix', '_advance_mana', '_advance_facilities', '_advance_guest_vitals', '_advance_staff_ai', '_advance_guest_ai_decisions', '_release_closed_business_queues', '_release_closed_business_request', '_complete_service_transaction'), 'scripts/facility/facility_service.gd': ( 'advance_one_minute', '_prepare_service_for_tick', '_complete_service', '_fail_service'), 'scripts/operations/operations_coordinator.gd': ('_process_minute_boundaries',), } # Condition -> (domain, authority refs, required engine measurements). # Existing safety/count thresholds are retained. Repair30's additional approved # order measurements still need the S5 raw adapter (see REPAIR30_STATUS.md). REQUIRED = { 'power_stop': ('停電', ('BASELINE:183', 'D05-P04-PAUSE-REASON'), {'charges': 0}), 'destruction_stop': ('破壊', ('D05-P04-DESTRUCTION',), {'charges': 0}), 'dependency_stop': ('閉店', ('D05-P04-DEPENDENCY',), {'charges': 0, 'progress_delta': 0, 'reservation_ids_preserved': True, 'service_id_preserved': True}), 'shift_completion': ('シフト交代', ('BASELINE:64', 'BASELINE:67', 'D05-P04-SHIFT-1800'), {'minute': 1080, 'completions': 1, 'charges': 1, 'completion_before_shift': True}), 'recovery_completion': ('課金', ('D05-P04-DEPENDENCY',), {'completions': 1, 'charges': 1}), 'same_minute_replay': ('課金', ('D05-P04-DEPENDENCY', 'D05-P04-SHIFT-1800'), {'completions': 1, 'charges': 1, 'duplicate_generations': 0, 'same_minute_replayed': True}), 'closure_release': ('閉店', ('D-034',), {'orphan_reservations': 0, 'released_once': True, 'started_service_continues': True}), 'exit_release': ('退出', ('D-034', 'D05-P04-EXIT'), {'orphan_reservations': 0, 'released_once': True}), 'patience_release': ('退出', ('D-034', 'D05-P04-EXIT'), {'orphan_reservations': 0, 'released_once': True}), 'paused_midnight': ('00:00決算', ('D05-P04-DEPENDENCY', 'BASELINE:71', 'BASELINE:72', 'D05-P04-MIDNIGHT-DAY'), {'orphan_reservations': 0, 'crossed_midnight_while_paused': True, 'stopped_charges': 0, 'completions': 1, 'charges': 1}), 'midnight_settlement': ('00:00決算', ('BASELINE:71', 'BASELINE:72', 'A01', 'D05-P04-MIDNIGHT-DAY'), {'settlements': 1, 'sales_payments': 1, 'payroll_payments': 1, 'upkeep_payments': 1, 'duty_payments': 1, 'nonzero_payroll': True}), 'midnight_replay': ('00:00決算', ('BASELINE:71', 'BASELINE:72', 'A01'), {'settlements': 1, 'same_minute_replayed': True, 'duplicate_payments': 0}), 'save_restore': ('00:00決算', ('BASELINE:71', 'BASELINE:72', 'A01'), {'settlements': 1, 'duplicate_payments': 0, 'formal_save_restored': True}), 'night_staff_absent': ('シフト交代', ('BASELINE:67', 'D05-P04-SHIFT-1800'), {'minute': 1081, 'state': 'WAITING_FOR_STAFF', 'night_staff_count': 0}), 'overtime_task': ('シフト交代', ('BASELINE:67',), {'interrupt_level': 0, 'is_overtime': True, 'duty_recorded_once': True}), } REACH_REFS = { 'EV': [ ('autoload/simulation_runtime.gd', '_execute_create_staff_task', '8b83aff45ead128b70931343271dbd03a1024b90238d1f46f93e4856290aa0d8', 10), ('scripts/staff/task_dispatcher.gd', '_reserve_pair', '48e2e593c117bd675236a165a294c9cc6233f87018010249cd8e675aff01fc3c', 6), ('autoload/simulation_runtime.gd', '_reserve_staff_task_items', '4f2add093de47cb08abfa45e9d2285f00d906c9bceadb2d98ffa42faae9ede67', 28), ('autoload/simulation_runtime.gd', '_advance_staff_task_travel', '53937aefeec533a5f16169bba4dd6a69b5246101d09de77ca95db7ca174c63ce', 2), ('scripts/transport/elevator_service.gd', 'set_powered', '925f4bf2d109421956fb2958981758d7d84a1ca586a9d4eb51572b8b9677cfba', 6)], 'transport': [ ('autoload/simulation_runtime.gd', '_on_staff_task_completed', '46daaa0197ce85c08883614112efb2d12a2cc0a02fe14ffec8314c73cbba61fc', 8), ('autoload/simulation_runtime.gd', '_handle_rescue_arrival', '11f8d43d0c089b95006009eddfc8bd3d4375a666b29ff19f36e1fa008f3946dd', 24), ('autoload/simulation_runtime.gd', '_advance_staff_rescue', '86bb31b40606fd624cc6b273dfdf7e3e4b3e1bc9997df825dd74136efb69a38f', 52)], } def require(condition, message): if not condition: raise ValueError(message) def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() def unique(pairs): result = {} for key, value in pairs: require(key not in result, 'duplicate JSON key: ' + key) result[key] = value return result def load(path): return json.loads(path.read_text(encoding='utf-8'), object_pairs_hook=unique, parse_constant=lambda x: require(False, 'nonfinite JSON: ' + x)) def safe(base, relative): require(isinstance(relative, str) and relative, 'invalid path') p = Path(relative) require(not p.is_absolute() and all(x not in {'.', '..'} for x in relative.split('/')), 'unsafe path: ' + relative) current = base for part in p.parts: current = current / part require(not current.is_symlink(), 'symlink: ' + relative) require(current.resolve().is_relative_to(base.resolve()) and current.is_file(), 'missing/outside file: ' + relative) return current def pins(mapping, base): require(isinstance(mapping, dict) and mapping, 'missing pin map') for name, digest in mapping.items(): require(isinstance(digest, str) and re.fullmatch('[0-9a-f]{64}', digest), 'invalid hash: ' + name) require(sha(safe(base, name)) == digest, 'stale pin: ' + name) def pointer(value, path): require(isinstance(path, str) and (path == '' or path.startswith('/')), 'invalid JSON pointer') for token in path.split('/')[1:]: require(not re.search('~(?![01])', token), 'invalid pointer escape') token = token.replace('~1', '/').replace('~0', '~') if isinstance(value, list): require(re.fullmatch('0|[1-9][0-9]*', token) is not None, 'invalid array index') value = value[int(token)] else: require(isinstance(value, dict) and token in value, 'missing pointer: ' + path) value = value[token] return value def literal(a, b): return type(a) is type(b) and json.dumps(a, sort_keys=True) == json.dumps(b, sort_keys=True) def git_bytes(*args): env = _GIT_ENV if _GIT_ENV is not None else git_environment() return subprocess.check_output(['git', '--no-replace-objects', *args], cwd=ROOT, env=env) def parse_blob_batch(stdout, oids): """Decode length-framed output without splitting binary payloads into lines.""" result = {} offset = 0 for oid in oids: end = stdout.find(b'\n', offset) require(end >= 0, 'truncated blob batch header') header = stdout[offset:end].split(b' ') require(len(header) == 3 and header[0] == oid.encode('ascii') and header[1] == b'blob' and re.fullmatch(rb'[0-9]+', header[2]) is not None, 'invalid blob batch header: ' + oid) size = int(header[2]) start = end + 1 end = start + size require(end < len(stdout) and stdout[end:end + 1] == b'\n', 'truncated blob batch payload or missing LF: ' + oid) result[oid] = stdout[start:end] offset = end + 1 require(offset == len(stdout), 'extra blob batch output') return result def read_blobs(oids): """One process per request; no cache can cross object namespaces.""" requested = list(oids) require(all(isinstance(oid, str) and re.fullmatch('[0-9a-f]{40}', oid) is not None for oid in requested), 'invalid full blob OID') requested = sorted(set(requested)) if not requested: return {} env = _GIT_ENV if _GIT_ENV is not None else git_environment() # Inherit stderr and raise on nonzero status, exactly as git_bytes does. stdout = subprocess.check_output(['git', '--no-replace-objects', 'cat-file', '--batch'], input=('\n'.join(requested) + '\n').encode('ascii'), cwd=ROOT, env=env) return parse_blob_batch(stdout, requested) def git_environment(): env = {k: v for k, v in os.environ.items() if not k.startswith('GIT_')} env.update(GIT_OPTIONAL_LOCKS='0', GIT_NO_REPLACE_OBJECTS='1', GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL=os.devnull) return env @contextmanager def isolated_objects(): """merge-tree writes only disposable objects; canonical refs/index stay untouched. Keep this context alive while reading the composed tree. No stage switch can bypass it. The alternate is only an object lookup path, never a write target. """ global _GIT_ENV require(_GIT_ENV is None, 'nested object store') objects = git_bytes('rev-parse', '--path-format=absolute', '--git-path', 'objects').decode().strip() build = Path(os.environ['BUILD_OUTPUT_DIR']).resolve(strict=True) with tempfile.TemporaryDirectory(prefix='p04-object-audit-', dir=build) as temporary: env = git_environment() env.update(GIT_OBJECT_DIRECTORY=temporary, GIT_ALTERNATE_OBJECT_DIRECTORIES=objects, GIT_OPTIONAL_LOCKS='0', GIT_NO_REPLACE_OBJECTS='1') _GIT_ENV = env try: yield finally: _GIT_ENV = None def tree_entries(tree): entries = {} for row in git_bytes('ls-tree', '-r', '-z', tree).split(b'\0'): if not row: continue metadata, encoded = row.split(b'\t', 1) mode, kind, oid = metadata.decode().split() path = encoded.decode('utf-8') require(kind == 'blob' and mode in {'100644', '100755'}, 'nonordinary tree entry: ' + path) require(path not in entries, 'duplicate tree path') entries[path] = (mode, oid) return entries def selected_path(path): parts = path.split('/') return (parts[0] in {'autoload', 'scripts', 'scenes', 'data', 'tests', 'assets', 'tools'} and len(parts) > 1 and parts[-1] != '.DS_Store' and '__pycache__' not in parts or path in {'project.godot', 'export_presets.cfg', 'Makefile', 'AGENTS.md'}) def allowed_path(path): return path == CHECKER_PATH or path.startswith(DOC_SCOPE) def product_manifest(entries): result = {} blobs = read_blobs(oid for path, (_, oid) in entries.items() if selected_path(path) and path != CHECKER_PATH) for path, (_, oid) in sorted(entries.items()): if selected_path(path) and path != CHECKER_PATH: data = blobs[oid] result[path] = {'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()} return result def selected_manifest(entries, require_checker=True): result = product_manifest(entries) if require_checker: require(CHECKER_PATH in entries, 'selected source missing checker') if CHECKER_PATH in entries: data = git_bytes('cat-file', 'blob', entries[CHECKER_PATH][1]) result[CHECKER_PATH] = {'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()} return result def manifest_digest(manifest): return hashlib.sha256(json.dumps(manifest, sort_keys=True, separators=(',', ':')).encode()).hexdigest() def supplied_file(directory, descriptor): """Read supplied material in place, never copy it or follow provenance paths.""" name = descriptor['path'] path = Path(name) if path.is_absolute(): support = directory / 'audit-support' roots = [support / n for n in ('p04-whole-closure-r01', 'native-first-current', 'p04-current-native-series-r01', 'p04-r17-full-inputs-r01')] fresh_path = support / 'p04-fresh-runs.json' if fresh_path.is_file(): fresh = load(fresh_path) require(fresh['protocol'] == 'p04-fresh-runs-v1', 'fresh material registry') roots.append(support / 'p04-fresh-context') for row in fresh['series'].values(): relative = Path(row['run']) require(not relative.is_absolute() and '..' not in relative.parts, 'fresh material run path') roots.append(directory / relative) matches = [root for root in roots if path.is_relative_to(root)] require(len(matches) == 1, 'unsupplied external material: ' + name) path = safe(matches[0], path.relative_to(matches[0]).as_posix()) else: path = safe(directory / 'p04-evidence', name) require(type(descriptor.get('bytes')) is int and path.stat().st_size == descriptor['bytes'] and sha(path) == descriptor.get('sha256'), 'supplied material pin: ' + name) return path def ancestor(older, newer): # merge-base's nonzero status is data only for the ordinary non-ancestor case. try: git_bytes('merge-base', '--is-ancestor', older, newer) return True except subprocess.CalledProcessError as exc: require(exc.returncode == 1, 'cannot determine ancestry') return False def submission_role(commit): require(ancestor(REGISTERED_COMMIT, commit) and not ancestor(MAIN_ANCHOR, commit), 'invalid submission ancestry') changed = git_bytes('diff', '--no-ext-diff', '--no-renames', '--name-only', '-z', REGISTERED_COMMIT, commit).split(b'\0') require(all(allowed_path(p.decode()) for p in changed if p), 'submission outside allowed_paths') def actual_checkout(): """Bind the executing modules and every worktree file to the actual HEAD.""" commit = git_bytes('rev-parse', '--verify', 'HEAD^{commit}').decode().strip() tree = git_bytes('rev-parse', 'HEAD^{tree}').decode().strip() entries = tree_entries(tree) require(not git_bytes('status', '--porcelain=v1', '--untracked-files=all', '--ignore-submodules=none'), 'checkout is not clean') require(not git_bytes('ls-files', '--others', '-z'), 'untracked/ignored substitution') blobs = read_blobs(oid for _, oid in entries.values()) for path, (mode, oid) in entries.items(): actual = safe(ROOT, path) require(actual.read_bytes() == blobs[oid], 'HEAD/tree/FS bytes: ' + path) require(('100755' if actual.stat().st_mode & 0o111 else '100644') == mode, 'HEAD/tree/FS mode: ' + path) # Also reject untracked directory symlinks, including ones ignored by Git. for parent, directories, files in os.walk(ROOT, followlinks=False): if Path(parent) == ROOT: directories[:] = [p for p in directories if p != '.git'] files = [p for p in files if p != '.git'] for name in directories + files: require(not (Path(parent) / name).is_symlink(), 'worktree symlink') require(Path(__file__).resolve() == safe(ROOT, CHECKER_PATH), 'checker module location') return commit, tree, entries def verify_roles(index, directory): """Infer roles from Git, never from an index or a caller's stage label.""" require(not ({'candidate_commit', 'candidate_tree', 'code_only', 'candidate', 'phase'} & index.keys()), 'candidate/code-only/phase declarations are not evidence') candidate = None commit, actual_tree, actual = actual_checkout() require(_GIT_ENV is not None, 'isolated object store required') if not ancestor(MAIN_ANCHOR, commit): submission_role(commit) submitted, main, role = commit, MAIN_ANCHOR, 'prospective composition' else: parents = git_bytes('show', '-s', '--format=%P', commit).decode().split() require(len(parents) == 2, 'integration/final requires a two-parent no-ff merge') main, submitted = parents require(main == MAIN_ANCHOR, 'integration main is not frozen M') submission_role(submitted) role = 'integration/final' composed = git_bytes('-c', 'rerere.enabled=false', 'merge-tree', '--write-tree', main, submitted).decode().strip() require(re.fullmatch('[0-9a-f]{40}', composed) is not None, 'conflicted composition') if 'fresh_run_bindings' in index: fresh = index['fresh_run_bindings'] require((fresh['M'], fresh['S'], fresh['Q_tree']) == (main, submitted, composed), 'fresh evidence does not bind recomputed Q') main_entries, patch, target = [tree_entries(t) for t in (main, submitted, composed)] anchor_entries = tree_entries(MAIN_ANCHOR) for path in main_entries.keys() | target.keys(): require(main_entries.get(path) == target.get(path) or allowed_path(path), 'composition changed game/unknown source: ' + path) # Metadata membership comes from the real main anchor, never a declared exclusion. metadata = sorted(p for p in anchor_entries if not selected_path(p)) for path in target: require(selected_path(path) or allowed_path(path) or path in metadata, 'unknown source hidden as metadata: ' + path) if role == 'integration/final': require(actual_tree == composed and actual == target, 'actual merge differs from composition') candidate = qualify_normal_candidate(commit, actual_tree, main, submitted, composed) require(target[CHECKER_PATH] == patch[CHECKER_PATH] == actual[CHECKER_PATH], 'checker blob differs from submission') for path in patch: if allowed_path(path): require(target.get(path) == patch[path], 'reviewed patch differs in merge: ' + path) binding = index['product_binding'] require(set(binding) == {'tested_product', 'runner_inputs'}, 'unknown product role/metadata declaration') tested = binding['tested_product'] require(tested == composed and git_bytes('cat-file', '-t', tested).strip() == b'tree', 'tested Q is not the genuine recomputed merge-tree') selection = git_bytes('show', MAIN_ANCHOR + ':tools/run_all_tests.py') require(hashlib.sha256(selection).hexdigest() == SELECTION_SHA, 'source selection authority') product = product_manifest(target) tested_entries = tree_entries(tested) expected = product_manifest(tested_entries) require(product == expected and product_manifest(main_entries) == expected and product_manifest(anchor_entries) == expected, 'complete selected native product mismatch') require(all(target[p][0] == main_entries[p][0] == tested_entries[p][0] for p in product), 'native product mode mismatch') selected_product = selected_manifest(target) require(set(selected_product) == set(product) | {CHECKER_PATH}, 'checker must be explicitly selected') require(index['source_map'] == {p: v['sha256'] for p, v in selected_product.items()}, 'stale pin: source map') runner_paths = set() for descriptor in binding['runner_inputs']: path = supplied_file(directory, descriptor) require(path not in runner_paths, 'duplicate runner input') runner_paths.add(path) native = load(path) before = native['source_before'] require(set(before) == set(selected_product) and all( {'bytes': v['bytes'], 'sha256': v['sha256']} == selected_product[p] for p, v in before.items()), 'runner complete source map mismatch') require(native['runner_sha256'] == product['tools/run_all_tests.py']['sha256'], 'runner source pin') overrides = native['QA_overrides'] require(len(overrides) == 1 and overrides[0].startswith('tests/') and overrides[0] in product, 'QA override membership') selected = {p: v['sha256'] for p, v in selected_product.items()} selected[overrides[0]] = native['QA']['sha256'] require(native['selected_source_map'] == selected, 'selected QA source map') needed = native.get('fixed_input_pins', []) + [native['QA']] require(set(descriptor['materials']) == {p['path'] for p in needed}, 'input material membership') for pin in needed: material = supplied_file(directory, descriptor['materials'][pin['path']]) require(material.stat().st_size == pin['bytes'] and sha(material) == pin['sha256'], 'input bytes') require(sha(supplied_file(directory, descriptor['wrapper'])) == native['wrapper_sha256'], 'wrapper bytes') require(runner_paths, 'runner inputs missing') require(index['versions'] == {'registered': REGISTERED_COMMIT, 'execution_origin': EXECUTION_ORIGIN, 'tested': tested}, 'evidence version roles') return composed, {'candidate_structure': candidate, 'role': role, 'actual_head': commit, 'actual_tree': actual_tree, 'submission': submitted, 'main': main, 'existing_metadata': metadata, 'manifest_sha256': manifest_digest(product), 'runner_paths': runner_paths, 'tested_product': tested, 'product_file_count': len(product), 'selected_file_count': len(selected_product)} def qualify_normal_candidate(commit, tree, main, submitted, composed): """Qualify structure only; Runner and Root verify provenance externally. The documentation roster is never consumed. Structure grants no provenance verification or original credit. """ require(tree == composed and main == MAIN_ANCHOR and commit != submitted, 'normal candidate identity mismatch') return {'structure_qualified': True, 'provenance_verified_by_checker': False, 'provenance_owner': 'actual Runner.prepare_integration/integrate and Root normal_c_join AFTER final before original credit', 'join': {'C': commit, 'S': submitted, 'M': main, 'C_tree': tree, 'Q': composed}} def function_bytes(data): """Top-level named func through next named func, preserving every raw byte. The prefix is an explicit non-function unit. Nested lambdas do not start a unit. Comments, whitespace, signatures and trailing declarations remain included; uncertain edits fail conservatively, never normalize as equal. """ matches = list(re.finditer(rb'(?m)^(?:static )?func ([A-Za-z_][A-Za-z_0-9]*)\(', data)) result = {'': data[:matches[0].start()] if matches else data} for i, match in enumerate(matches): name = match[1].decode('ascii') require(name not in result, 'ambiguous function: ' + name) result[name] = data[match.start():matches[i + 1].start() if i + 1 < len(matches) else len(data)] return result def resolve_anchor(data, function, digest, line_count): body = function_bytes(data).get(function) require(body is not None, 'reachability function missing: ' + function) lines = body.splitlines(keepends=True) hits = [b''.join(lines[i:i + line_count]) for i in range(len(lines) - line_count + 1) if hashlib.sha256(b''.join(lines[i:i + line_count])).hexdigest() == digest] require(len(hits) == 1, 'reachability anchor missing/ambiguous: ' + function) return hits[0] def source_delta(base, head): """Compute changed function units from Git, including added/deleted files. Only game autoload/scripts GDScript belongs to this applicability scope. The complete tested map still pins every supplied dependency separately. """ paths = git_bytes('diff', '--no-ext-diff', '--no-renames', '--name-only', '-z', base + '..' + head, '--', 'autoload/', 'scripts/').split(b'\0') base_paths = set(git_bytes('ls-tree', '-r', '--name-only', '-z', base).split(b'\0')) head_paths = set(git_bytes('ls-tree', '-r', '--name-only', '-z', head).split(b'\0')) delta = [] for encoded in sorted(p for p in paths if p.endswith(b'.gd')): path = encoded.decode('utf-8') old = function_bytes(git_bytes('show', base + ':' + path)) if encoded in base_paths else {} new = function_bytes(git_bytes('show', head + ':' + path)) if encoded in head_paths else {} for name in sorted(old.keys() | new.keys()): if old.get(name) != new.get(name): delta.append({'path': path, 'function': name}) return delta def boundary_conditions(path, function): # Conservative coverage: new/changed Runtime setup can reach every boundary. if path == 'autoload/simulation_runtime.gd': if any(word in function for word in ('relationship', 'restore', 'snapshot', 'save')): return {'save_restore'} return set(REQUIRED) | {'EV'} if '/relationship/' in path or '/save/' in path: return {'save_restore'} if any(part in path for part in ('/world/', '/transport/', '/staff/')): return {'exit_release', 'closure_release', 'shift_completion', 'EV', 'save_restore'} # No inferred non-impact exemption for previously unseen dependencies. return set(REQUIRED) | {'EV'} def mandatory_conditions(index): conditions = index.get('conditions') require(isinstance(conditions, dict), 'mandatory conditions must be a dictionary') missing = sorted(set(REQUIRED) - set(conditions)) unknown = sorted(set(conditions) - set(REQUIRED)) require(not missing and not unknown, 'mandatory conditions missing: ' + ', '.join(missing) + '; unknown: ' + ', '.join(unknown)) return conditions def applicability(index, spec, source, head): mandatory_conditions(index) tested = index.get('versions', {}).get('tested') require(isinstance(tested, str) and tree_entries(tested) and product_manifest(tree_entries(tested)) == product_manifest(tree_entries(head)), 'wrong tested revision') registered = spec['current_source_pins'] require(index.get('registered_pins') == registered, 'registered pin declaration') for path, digest in registered.items(): require(hashlib.sha256(git_bytes('show', REGISTERED_COMMIT + ':' + path)).hexdigest() == digest, 'registered Git pin: ' + path) # head is the independently recomputed product tree, not the checker HEAD. game_paths = git_bytes('ls-tree', '-r', '--name-only', '-z', head, '--', 'autoload/', 'scripts/').split(b'\0') for encoded in game_paths: if not encoded.endswith(b'.gd'): continue path = encoded.decode('utf-8') require(path in source and source[path] == hashlib.sha256(git_bytes('show', head + ':' + path)).hexdigest(), 'tested game differs from Git: ' + path) delta = source_delta(REGISTERED_COMMIT, head) require(index.get('applicability_delta') == delta, 'git diff/applicability_delta mismatch') for path, names in ORDER_FUNCTIONS.items(): old = function_bytes(git_bytes('show', REGISTERED_COMMIT + ':' + path)) new = function_bytes(git_bytes('show', head + ':' + path)) for name in names: require(name in old and name in new and old[name] == new[name], 'stale ordering function: ' + path + ':' + name) bindings = index.get('delta_cases') require(isinstance(bindings, dict) and set(bindings) == {r['path'] + ':' + r['function'] for r in delta}, 'delta case bindings') for row in delta: key = row['path'] + ':' + row['function'] required = boundary_conditions(row['path'], row['function']) cases = bindings[key] require(isinstance(cases, dict) and set(cases) == required, 'boundary delta coverage: ' + key) for cid, case_id in cases.items(): declaration = index['reachability']['EV'] if cid == 'EV' else index['conditions'][cid] require(case_id == declaration.get('case_id'), 'boundary delta engine case: ' + key) def owner_authority(ref, entry, path): """Each decision is bound to its own immutable, path-qualified record.""" require(ref in OWNER_BINDINGS, 'unknown owner authority: ' + ref) relative, digest = OWNER_BINDINGS[ref] require(entry.get('status') == 'approved', 'pending/missing authority: ' + ref) require(entry.get('source') == relative, 'owner approval path mismatch: ' + ref) require(sha(path) == digest, 'owner approval hash mismatch: ' + ref) owner = load(path) require(owner.get('origin') == 'direct_user_reply' and owner.get('formal_acceptance') is False, 'owner approval scope') require(ref in {v['id'] for v in owner['approved_choices']}, 'owner approval condition') def coverage(index, raw, source, product_tree): authority = index['authority'] for ref in {r for _, refs, _ in REQUIRED.values() for r in refs}: entry = authority.get(ref, {}) require(entry.get('status') in {'original', 'approved'}, 'pending/missing authority: ' + ref) aid = entry.get('artifact_id') require(aid in raw and index['artifacts'][aid]['type'] == 'authority', 'raw authority: ' + ref) if ref.startswith('D05-P04-'): owner_authority(ref, entry, raw[aid]) else: path = ('docs/traceability/original/BASELINE.txt' if ref.startswith('BASELINE:') else 'docs/DESIGN_DECISIONS.md' if ref == 'D-034' else 'docs/plans/original-completion-2026-10-02/execution/p04-01-input-r01/ORDER_CONTRACT_DRAFT.md') require(sha(raw[aid]) == sha(safe(ROOT, path)), 'authority document mismatch: ' + ref) conditions = mandatory_conditions(index) for cid, (domain_name, refs, expected) in REQUIRED.items(): condition = conditions[cid] require(condition.get('authority_refs') == list(refs), 'condition authority: ' + cid) require(condition.get('nativeWindows') == 'pending', 'nativeWindows condition: ' + cid) domain = index['domains'][domain_name] require(set(refs) <= set(domain['authority_refs']), 'domain authority: ' + cid) cases = [c for c in domain['cases'] if c['id'] == condition.get('case_id')] require(len(cases) == 1, 'mandatory engine case: ' + cid) case = cases[0] require(case.get('raw_origin') == 'engine', 'case origin: ' + cid) observation = load(raw[case['after']]) measurements = observation.get('conditions', {}).get(cid) require(isinstance(measurements, dict), 'missing engine measurements: ' + cid) for metric, value in expected.items(): require(literal(measurements.get(metric), value), cid + ': measurement ' + metric) require(any(p.get('case_id') == case['id'] and p.get('artifact_id') == case['after'] and p.get('json_pointer') == '/conditions/' + cid + '/' + metric and p.get('operator') == 'eq' and literal(p.get('expected'), value) and p.get('authority_ref') in refs for p in domain['invariants']), cid + ': missing mandatory invariant ' + metric) if cid == 'save_restore': writer = load(raw[case['receipt']]) reader_id = condition.get('restore_receipt') require(reader_id in raw, 'missing restore receipt') reader = load(raw[reader_id]) require(type(reader.get('start_pid')) is int and reader['start_pid'] > 0 and reader['start_pid'] != writer['start_pid'] and reader.get('end_pid') == reader['start_pid'] and type(reader.get('exit_code')) is int and reader['exit_code'] == 0 and reader.get('engine_ended') is True, 'separate PID restore receipt') require(reader.get('source_map') == source and reader.get('environment') == index['environment'], 'restore applicability') require(reader.get('case_id') == case['id'] and reader.get('artifact_hashes') == writer.get('artifact_hashes'), 'restore case artifact binding') save_id = condition.get('save_artifact') require(save_id in raw and index['artifacts'][save_id]['type'] == 'save', 'formal save artifact') require(writer.get('saved_sha256') == sha(raw[save_id]) == reader.get('restored_sha256'), 'save/restore identity') require(reader.get('stdout') in raw and reader.get('native_log') in raw, 'restore logs') reachability = index.get('reachability') require(isinstance(reachability, dict) and set(reachability) == {'EV', 'transport'}, 'reachability required') for name, decision in reachability.items(): refs = decision.get('source_refs') require(isinstance(refs, list) and refs and isinstance(decision.get('reason'), str) and decision['reason'].strip(), 'unsupported reach=false: ' + name) for ref in refs: require(ref.get('path') in source and ref.get('sha256') == source[ref['path']], 'reachability source pin') identity = (ref['path'], ref.get('function'), ref.get('anchor_sha256'), ref.get('line_count')) require(identity in REACH_REFS[name], 'unknown reachability anchor') resolve_anchor(git_bytes('show', product_tree + ':' + ref['path']), *identity[1:]) require(len(refs) == len(REACH_REFS[name]) and set(REACH_REFS[name]) == { (r['path'], r['function'], r['anchor_sha256'], r['line_count']) for r in refs}, 'missing static reachability chain: ' + name) # Staff dispatch reserves task items before travel. A guest-only exclusion # must not hide this reachable EV reservation path. if name == 'EV': require(decision.get('reach') is True, 'EV staff reservation path is reachable') domain = index['domains']['停電'] cases = [c for c in domain['cases'] if c['id'] == decision.get('case_id')] require(len(cases) == 1 and cases[0].get('raw_origin') == 'engine', 'EV engine case') case = cases[0] ev = load(raw[case['after']]).get('ev', {}) require(ev.get('reserved_before_travel') is True and ev.get('power_stopped') is True, 'EV reserved staff power-stop scenario') require(any(p.get('case_id') == case['id'] and p.get('artifact_id') == case['after'] and p.get('json_pointer') == '/ev/orphan_reservations' and p.get('operator') == 'eq' and literal(p.get('expected'), 0) for p in domain['invariants']), 'EV orphan invariant') else: require(decision.get('reach') is False, 'transport reservation/payment exclusion') NATIVE_RUNS = { 'actual-native': ('import', 'parse', 'repro'), 'dependency-midnight-native': ('import', 'parse', 'repro'), 'dependency-formal-native': ('import', 'parse', 'writer', 'reader'), } SAVE_POINTS = ('before', 'midnight', 'paused', 'after', 'next') SAVE_MEMBERS = {'journal.log', 'manifest.json', 'thumbnail.png', 'world.bin'} def native_file(base, name, pin): path = safe(base, name) require(type(pin.get('bytes')) is int and path.stat().st_size == pin['bytes'], 'native bytes: ' + name) require(sha(path) == pin.get('sha256'), 'native SHA: ' + name) return path def native_cleanup(value): require(value.get('engine_ended') is True and value.get('errors') == [] and literal(value.get('actual_process_exit_code'), 0), 'native cleanup failed') attempts = value.get('attempts') require(isinstance(attempts, list) and attempts, 'native cleanup attempts missing') for attempt in attempts: require(attempt.get('errors') == [] and attempt.get('kill_attempted') is False and attempt.get('wait_completed') is True and literal(attempt.get('wait_returned_code'), 0), 'native cleanup abnormal wait') def native_ownership(base): provenance = load(safe(base, 'OWNERSHIP_RAW_PROVENANCE.json'))['copied'] targets = set() for row in provenance: require(row['target'] not in targets, 'duplicate native provenance target') targets.add(row['target']) native_file(base, row['target'], row) pids = {} for run, phases in NATIVE_RUNS.items(): execution = load(safe(base, run + '/execution.json')) inputs = load(safe(base, run + '/runner-input.json')) require([p['phase'] for p in execution['phases']] == list(phases), 'native phase membership') run_pids = [] cleanups = [] for phase, phase_receipt in zip(phases, execution['phases']): prefix = run + '/' + phase + '/' names = ('started.json', 'process.json', 'ended.json', 'outer-owned-finally.json', 'adapter-owned-cleanup.json', 'stdout.log', 'stderr.log', 'engine-native.log') require({prefix + n for n in names} <= targets, 'native provenance lifecycle missing') start, process, end, outer, cleanup = [load(safe(base, prefix + n)) for n in names[:5]] pid = start['actual_pid'] require(type(pid) is int and pid > 0 and pid not in pids.values(), 'native PID reused') require(start['pid'] == process['pid'] == end['pid'] == end['actual_pid'] == pid, 'native PID chain') require(start['event'] == 'ENGINE_STARTED' and end['event'] == 'ENGINE_ENDED' and start['owner_pid'] == end['owner_pid'] == inputs['owner_runner_pid'] and start['command'] == start['argv'] == process['argv'] == end['argv'] == end['command'] and start['monotonic_ns'] < end['monotonic_ns'], 'native lifecycle identity') require(process['actual_start_binding'] == start and process['actual_end_binding'] == end, 'native process bindings') require(start['current_process_identity']['pid'] == pid and start['current_process_identity']['ppid'] == start['owner_pid'], 'native owner identity') require(load(safe(base, run + '/' + phase + '-receipt.json')) == phase_receipt == load(safe(base, run + '/' + phase + '-actual-phase.json')), 'native phase receipt binding') for record in (process, end, phase_receipt): require(literal(record.get('exit_code'), 0), 'native abnormal exit') require(end.get('engine_ended') is True and end.get('timed_out') is False and process.get('timed_out') is False and end.get('lifecycle_error') is None and process.get('lifecycle_error') is None and literal(end.get('actual_process_exit_code'), 0) and phase_receipt.get('engine_ended') is True, 'native abnormal end') require(phase_receipt['actual_pid'] == pid and phase_receipt['native_process'] == process, 'native execution process binding') require(outer['actual_captured_pids'] == [pid] and outer['outer_cleanup_errors'] == [] and outer.get('all_captured_ended') is True and outer.get('adapter_exception') is None and len(outer['cleanup']) == 1 and outer['cleanup'][0]['pid'] == pid and outer['cleanup'][0]['outer_cleanup_errors'] == [] and outer['cleanup'][0]['cleanup'] == cleanup == process['adapter_owned_cleanup'], 'native owned cleanup binding') native_cleanup(cleanup) for stream in ('stdout', 'stderr'): require(sha(safe(base, prefix + stream + '.log')) == process[stream + '_sha256'], 'native stream hash') for field in ('stdout', 'stderr'): require(safe(base, prefix + field + '.log').is_file(), 'native log missing') native_file(base, prefix + 'engine-native.log', phase_receipt['raw_native_log']) pids[run + '/' + phase] = pid run_pids.append(pid) cleanups.extend(outer['cleanup']) require(execution['actual_captured_PIDs'] == run_pids and execution['cleanup'] == cleanups and execution['cleanup_errors'] == [], 'native execution cleanup membership') formal = 'dependency-formal-native/' records = [load(safe(base, formal + 'restart-evidence/' + role + '.json')) for role in ('writer', 'reader')] require(records[0]['profile'] != records[1]['profile'], 'native writer reader profile reused') for role, record in zip(('writer', 'reader'), records): require(record['pid'] == pids[formal + role], 'native writer reader PID binding') profile = load(safe(base, formal + role + '-profile.json')) phase = load(safe(base, formal + role + '-receipt.json')) require(record['profile'] == profile['actual_user_data_dir'] == profile['expected_user_data_dir'] == phase['controlled_environment']['THE_TREE_PACK_PROFILE_EXPECTED'], 'native profile binding') require(record['slots'] == {p: record['profile'] + '/saves/p04_boundary/' + p for p in SAVE_POINTS}, 'native formal profile slots') return pids def native_applicability(base): """Production evidence identity, separate from A/B checkout applicability.""" app = load(safe(base, 'POST_P18_CURRENT_SOURCE_APPLICABILITY.json')) production = app['production_commit'] require(re.fullmatch('[0-9a-f]{40}', production) is not None, 'native production commit') registered = load(SPEC)['current_source_pins'] pin_rows = app['registered_source_pins'] require(len(pin_rows) == len(registered) and {r['path'] for r in pin_rows} == set(registered), 'native current_source_pins membership') current = {} for row in pin_rows: path = row['path'] current[path] = hashlib.sha256(git_bytes('show', production + ':' + path)).hexdigest() require(row['registered_sha256'] == registered[path] and row['production_sha256'] == current[path], 'native current_source_pins mismatch') rows = app['ordering_functions'] require(len(rows) == sum(map(len, ORDER_FUNCTIONS.values())) and {(r['path'], r['function']) for r in rows} == {(p, f) for p, fs in ORDER_FUNCTIONS.items() for f in fs}, 'native ordering membership') for path, functions in ORDER_FUNCTIONS.items(): old = function_bytes(git_bytes('show', REGISTERED_COMMIT + ':' + path)) new = function_bytes(git_bytes('show', production + ':' + path)) for function in functions: row = next(r for r in rows if (r['path'], r['function']) == (path, function)) body = new[function] require(old[function] == body and hashlib.sha256(body).hexdigest() == row['production_body_sha256'] == row['registered_body_sha256'], 'native ordering bytes') for run, result in zip(NATIVE_RUNS, ('SHIFT_CURRENT', 'DEPENDENCY_MIDNIGHT', 'DEPENDENCY_FORMAL')): summary = load(safe(base, 'ROOT_APPROVED_' + result + '_RESULT.json')) require(summary['production_commit'] == production, 'native production commit mismatch') inputs = load(safe(base, run + '/runner-input.json')) for path, digest in current.items(): require(inputs['source_before'][path]['sha256'] == digest == inputs['selected_source_map'][path], 'native input source pin mismatch') return production def native_saves(base): prefix = 'dependency-formal-native/' execution = load(safe(base, prefix + 'execution.json')) before, after = execution['formal_before'], execution['formal_after'] require(set(before) == set(after) == set(SAVE_POINTS) and before == after, 'native original members changed') summary = load(safe(base, 'ROOT_APPROVED_DEPENDENCY_FORMAL_RESULT.json')) members = summary['formal_member_pins'] require(len(members) == 20 and {(r['point'], r['member']) for r in members} == {(p, m) for p in SAVE_POINTS for m in SAVE_MEMBERS}, 'native formal member membership') for point in SAVE_POINTS: require(set(before[point]) == SAVE_MEMBERS, 'native original member missing') for member in SAVE_MEMBERS: for role in ('writer', 'reader'): path = native_file(base, prefix + 'formal-packages/' + role + '/' + point + '/' + member, before[point][member]) row = next(r for r in members if (r['point'], r['member']) == (point, member)) require(sha(path) == row['sha256'], 'native formal member pin') typed = summary['typed_outputs_14_sha_match_original_native'] expected = {role + '-' + p + suffix + '.var': p for p in SAVE_POINTS for role, suffix in (('writer', '-persisted'), ('reader', '-loaded'))} expected.update({'reader-replayed-' + p + '.var': p for p in SAVE_POINTS if p != 'before'}) require(len(typed) == len(expected) and {Path(r['output']).name for r in typed} == set(expected), 'native typed member membership') inputs = load(safe(base, prefix + 'runner-input.json'))['fixed_input_pins'] for row in typed: name = Path(row['output']).name original = 'midnight-' + expected[name] + '.var' require(Path(row['original']).name == original, 'native typed original identity') source = safe(base, 'dependency-midnight-native/order-evidence/' + original) output = safe(base, prefix + 'restart-evidence/' + name) require(sha(source) == sha(output) == row['sha256'], 'native typed SHA mismatch') pin = [p for p in inputs if p['path'] == row['original']] require(len(pin) == 1, 'native typed input missing') native_file(base, 'dependency-midnight-native/order-evidence/' + original, pin[0]) def receipt_rows(receipts): """Keep multiplicity: sets must never hide duplicated ledger rows.""" require(isinstance(receipts, list), 'native receipts missing') ids, rows = set(), [] for receipt in receipts: require(receipt['id'] not in ids and receipt['rows'], 'native duplicate/empty receipt') ids.add(receipt['id']) for row in receipt['rows']: require(row['transaction_id'] == receipt['id'] and type(row['amount']) is int and row['amount'] >= 0, 'native receipt transaction') require(row not in rows, 'native duplicate receipt row') rows.append(row) return rows def native_ledger(base): prefix = 'dependency-midnight-native/order-evidence/' observations = load(safe(base, prefix + 'observations.json')) series = observations['observations'] require(len(series) == len(SAVE_POINTS), 'native observation series missing') initial = series[0] service = initial['service'] require(service['state'] == 'ACTIVE' and service['elapsed_minutes'] == service['duration_minutes'] - 1, 'native initial service progress') identity = ('service_id', 'request_id', 'facility_id', 'guest_ids', 'inventory_reservations', 'price_snapshot') require(service['inventory_reservations'] and len(service['guest_ids']) == 1, 'native service identity missing') guest_id = service['guest_ids'][0] account = 'GUEST_WALLET:' + guest_id final_state = load(safe(base, prefix + 'midnight-after.json'))['systems']['runtime'] links = [r for r in final_state['economy']['idempotency'] if r['key'] == 'service.payment.' + service['service_id']] require(len(links) == 1, 'native service idempotency missing/duplicate') transaction = links[0]['transaction_id'] previous_rows = [] states = [] for position, (point, observation) in enumerate(zip(SAVE_POINTS, series)): snapshot = load(safe(base, prefix + 'midnight-' + point + '.json')) state = snapshot['systems']['runtime'] states.append(state) facilities = [f for f in state['facilities']['instances'] if f['instance_id'] == service['facility_id']] require(len(facilities) == 1, 'native facility identity') facility = facilities[0] if position == 0: definition_id = facility['definition_id'] require(re.fullmatch('[a-zA-Z0-9_]+', definition_id) is not None, 'native definition identity') definition_path = 'data/facilities/' + definition_id + '.tres' production = load(safe(base, 'POST_P18_CURRENT_SOURCE_APPLICABILITY.json'))['production_commit'] definition = git_bytes('show', production + ':' + definition_path) inputs = load(safe(base, 'dependency-midnight-native/runner-input.json')) require(inputs['selected_source_map'][definition_path] == inputs['source_before'][definition_path]['sha256'] == hashlib.sha256(definition).hexdigest(), 'native original definition pin') for field, value in (('base_price', service['price_snapshot']), ('base_service_minutes', service['duration_minutes'])): matches = re.findall(rb'(?m)^' + field.encode() + rb' = ([0-9]+)$', definition) require(len(matches) == 1 and int(matches[0]) == value == facility[field], 'native original definition value') current = observation['service'] require(all(current[k] == service[k] for k in identity), 'native service identity changed') require(observation['minute'] == initial['minute'] + position == snapshot['clock']['absolute_minute'], 'native minute series') require(current in state['facilities']['service_history'] or any( current in f['active_services'] for f in state['facilities']['instances']), 'native service snapshot binding') require(observation['reservations'] == state['inventory'], 'native reservation snapshot binding') rows = receipt_rows(observation['receipts']) ledger = state['economy']['entries'] initial_ledger = states[0]['economy']['entries'] require(ledger[:len(initial_ledger)] == initial_ledger and ledger[len(initial_ledger):] == rows, 'native ledger receipt delta') require(rows[:len(previous_rows)] == previous_rows, 'native receipt history changed') wallet_delta = sum(r['amount'] * ((r['debit_account'] == account) - (r['credit_account'] == account)) for r in rows) require(observation['wallet'] == initial['wallet'] + wallet_delta, 'native wallet delta') require(observation['wallet'] == sum(r['amount'] * ((r['debit_account'] == account) - (r['credit_account'] == account)) for r in ledger), 'native ledger wallet balance') if observation['guest']: require(observation['guest']['wallet_gold'] == observation['wallet'] and observation['guest']['guest_id'] == guest_id, 'native guest wallet binding') else: require(point == 'next', 'native guest missing before completion') sales = [r for r in rows if r['category'] == 'SERVICE_SALE'] paused = current['state'] == 'PAUSED' if position in (1, 2): require(paused and current['interruption_reason'] == 'RELATED_FACILITY_INACTIVE' and current['elapsed_minutes'] == service['elapsed_minutes'] and not sales, 'native PAUSED sale/progress/reason') elif position >= 3: require(current['state'] == 'COMPLETED' and len(sales) == 1 and current['elapsed_minutes'] == current['duration_minutes'], 'native completion sale count') sale = sales[0] require(sale['transaction_id'] == transaction and sale['credit_account'] == account and sale['source_entity_id'] == service['facility_id'] and sale['debit_account'] == 'FACILITY_REVENUE:' + service['facility_id'] and sale['amount'] == service['price_snapshot'] and sale['absolute_minute'] == current['completed_at'], 'native sale service join') reservations = observation['reservations']['reservations'] require(len(reservations) == len(service['inventory_reservations']) and {r['reservation_id'] for r in reservations} == set(service['inventory_reservations']), 'native orphan reservation membership') requests = state['facilities']['requests'] matching_requests = [r for r in requests if r['request_id'] == service['request_id']] require(len(matching_requests) == (1 if position < 3 else 0), 'native request series') if matching_requests: request = matching_requests[0] require(request['facility_id'] == service['facility_id'] and request['guest_ids'] == service['guest_ids'] and request['quoted_price'] == service['price_snapshot'] and request['reserved_inventory_ids'] == service['inventory_reservations'], 'native request service join') for reservation in reservations: require(reservation['requester_id'] == service['request_id'] and reservation['owner_inventory_id'] == 'facility.inventory:' + service['facility_id'] and reservation['state'] == ('ACTIVE' if position < 3 else 'CONSUMED'), 'native orphan reservation relation') previous_rows = rows require(receipt_rows(observations['first_receipts']) == previous_rows == receipt_rows(observations['replay_receipts']), 'native midnight replay receipts') reader = load(safe(base, 'dependency-formal-native/restart-evidence/reader.json')) require(len(reader['results']) == 1 and receipt_rows(reader['results'][0]['receipts']) == previous_rows, 'native reader receipt delta') sweep = [r for r in previous_rows if r['category'] == 'SERVICE_CASH_SWEEP'] require(len(sweep) == 1 and sweep[0]['transaction_id'] == transaction and sweep[0]['amount'] == service['price_snapshot'] and sweep[0]['credit_account'] == 'FACILITY_REVENUE:' + service['facility_id'] and sweep[0]['debit_account'] == 'PLAYER_CASH', 'native sale cash sweep') maintenance = [r for r in previous_rows if r['category'].startswith('MAINTENANCE_')] accruals = [r for r in maintenance if r['category'] == 'MAINTENANCE_ACCRUAL'] payments = [r for r in maintenance if r['category'] == 'MAINTENANCE_PAYMENT'] require(accruals and len(accruals) == len(payments), 'native maintenance pairs') for accrual in accruals: matches = [r for r in payments if r['transaction_id'] == accrual['transaction_id'] and r['source_entity_id'] == accrual['source_entity_id'] and r['amount'] == accrual['amount'] and r['absolute_minute'] == series[1]['minute'] == accrual['absolute_minute'] + 1] require(len(matches) == 1, 'native maintenance accrual/payment join') return {'service_id': service['service_id'], 'transaction_id': transaction, 'receipt_rows': len(previous_rows), 'formal_original_credit': 0} def verify_native(base): """Finite C-layer check, never a replacement for the 15+EV whole gate.""" for row in load(safe(base, 'SOURCE_MANIFEST.json'))['files']: native_file(base, row['copy'], row) production = native_applicability(base) pids = native_ownership(base) native_saves(base) result = native_ledger(base) return dict(result, production_commit=production, owned_processes=len(pids), verified_conditions=['dependency_stop', 'recovery_completion', 'paused_midnight']) def verify_shift_completion(base): """C's first 18:00 transaction, never the paid-midnight 1442 sale.""" for row in load(safe(base, 'SOURCE_MANIFEST.json'))['files']: native_file(base, row['copy'], row) tested = native_applicability(base) pids = native_ownership(base) shift_pids = [pids['actual-native/' + phase] for phase in NATIVE_RUNS['actual-native']] require(len(shift_pids) == len(set(shift_pids)) == 3, 'shift ended PID identity') inputs = load(safe(base, 'actual-native/runner-input.json')) expected = selected_manifest(tree_entries(tested), require_checker=False) require({p: {'bytes': v['bytes'], 'sha256': v['sha256']} for p, v in inputs['source_before'].items()} == expected, 'shift complete native source') prefix = 'actual-native/order-evidence/' observations = load(safe(base, prefix + 'observations.json')) labels = ['first-before-17-59', 'first-18-00-before-facility', 'first-18-00-after-facility', 'first-18-00-after-staff', 'first-18-00-tick-complete', 'first-18-01-tick-complete'] points = [] previous_ledger = None service_id = None paid_rows = None original_service = None completed_service = None transaction_id = None for i, label in enumerate(labels): rows = [r for r in observations['observations'] if r['label'] == label] require(len(rows) == 1, 'shift phase membership') observation = rows[0] snapshot = load(safe(base, prefix + label + '.json')) minute = [1079, 1080, 1080, 1080, 1080, 1081][i] require(snapshot['clock']['absolute_minute'] == observation['minute'] == minute, 'shift phase clock') state = snapshot['systems']['runtime'] service = observation['service'] if i == 0: service_id = service['service_id'] original_service = service identity = ('service_id', 'request_id', 'facility_id', 'guest_ids', 'inventory_reservations', 'price_snapshot', 'duration_minutes', 'started_at', 'staff_ids') require(all(service[k] == original_service[k] for k in identity), 'shift original service identity') require(service['service_id'] == service_id and service['elapsed_minutes'] == (44 if i < 2 else 45) and service['duration_minutes'] == 45 and service['price_snapshot'] == 25, 'shift service identity/progress') candidates = state['facilities']['service_history'] + [s for f in state['facilities']['instances'] for s in f['active_services']] require([s for s in candidates if s['service_id'] == service_id] == [service], 'shift raw service join') require(observation['reservations'] == state['inventory'], 'shift inventory join') reserved = [r for r in state['inventory']['reservations'] if r['reservation_id'] in service['inventory_reservations']] require(reserved and len(reserved) == len(service['inventory_reservations']) and all( r['requester_id'] == service['request_id'] and r['state'] == ('ACTIVE' if i < 2 else 'CONSUMED') and r['terminal_at'] == (-1 if i < 2 else 1080) for r in reserved), 'shift reservation join') day2 = ['p04.order.server.a', 'p04.order.server.b'] require(observation['active_staff'] == (day2 if i < 3 else []), 'shift DAY2 before/after handover') facility = [f for f in state['facilities']['instances'] if f['instance_id'] == service['facility_id']] require(len(facility) == 1 and facility[0]['active_staff_ids'] == observation['active_staff'], 'shift full-state active staff join') require([w['staff']['staff_id'] for w in observation['workers']] == day2, 'shift worker identity set') for worker in observation['workers']: require(worker['staff'] in state['staff']['staff'], 'shift worker raw join') if i < 3: require(worker['shift'] in state['staff']['shifts'] and worker['is_present'] is True and worker['shift']['kind'] == 'DAY' and worker['shift']['end_at'] == 1080 and worker['shift']['actual_ended_at'] == -1, 'shift original DAY worker present') else: ended = [r for r in state['staff']['shifts'] if r['staff_id'] == worker['staff']['staff_id']] require(worker['shift'] == {} and worker['is_present'] is False and worker['staff']['state'] == 'OFF_DUTY' and len(ended) == 1 and ended[0]['actual_ended_at'] == ended[0]['end_at'] == 1080, 'shift normal handover history') ledger = state['economy']['entries'] sale = [r for r in ledger if r['category'] == 'SERVICE_SALE'] if i < 2: require(sale == [] and observation['receipts'] == [], 'shift precompletion charge') else: require(service['state'] == 'COMPLETED' and service['completed_at'] == 1080 and len(sale) == 1, 'shift single completion') transaction = sale[0] if transaction_id is None: transaction_id = transaction['transaction_id'] require(transaction['transaction_id'] == transaction_id and transaction['amount'] == service['price_snapshot'] and transaction['absolute_minute'] == 1080 and transaction['source_entity_id'] == service['facility_id'] and transaction['credit_account'] == 'GUEST_WALLET:' + service['guest_ids'][0], 'shift sale identity') receipts = receipt_rows(observation['receipts']) require(receipts == [r for r in ledger if r['transaction_id'] == transaction_id], 'shift receipt raw join') require([r for r in state['economy']['idempotency'] if r['key'] == 'service.payment.' + service_id] == [{'key': 'service.payment.' + service_id, 'transaction_id': transaction_id}], 'shift idempotency join') if i == 2: completed_service = service paid_rows = observation['receipts'] require(paid_rows == observations['first_receipts'] and paid_rows[0]['staff_at_receipt'] == day2, 'shift receipt precedes handover') else: require(service == completed_service and ledger == previous_ledger and observation['receipts'] == paid_rows, 'shift extra charge/completion') previous_ledger = ledger points.append(label) return {'verified_conditions': ['shift_completion'], 'phases': points, 'transaction_id': transaction_id} def combine_historical_native_series(directory, native): """Independent R composition; rerun raw checks, never trust condition scalars. The C verifier and its fixtures remain unchanged. This function does not establish candidate applicability, and cannot grant a successful whole gate. """ import importlib.util module_path = ROOT / DOC_SCOPE / 'p04_native_series.py' module_spec = importlib.util.spec_from_file_location('p04_series', module_path) series = importlib.util.module_from_spec(module_spec) module_spec.loader.exec_module(series) c_base = directory / 'audit-support/native-first-current' require(native == verify_native(c_base), 'C result must be recomputed from raw') base = directory / 'audit-support/p04-raw-collect-r22' manifest_path = safe(base, 'MANIFEST.json') manifest = load(manifest_path) require(manifest['protocol'] == 'p04-native-series-collection-v1' and manifest['relocation_manifest_sha256'] == series.MANIFEST_SHA256, 'series relocation binding') require(set(manifest['series']) == set(series.RUNS), 'series manifest membership') shift = verify_shift_completion(c_base) verified = set(native['verified_conditions']) | set(shift['verified_conditions']) bindings = {} for name in series.RUNS: binding = manifest['series'][name] path = safe(base, binding['path']) require(path.name == name + '.json' and sha(path) == binding['sha256'] and path.stat().st_size == binding['bytes'] and binding['relocation_manifest_sha256'] == series.MANIFEST_SHA256, 'series manifest SHA binding') recorded = load(path) require(recorded['verifier_sha256'] == sha(module_path), 'series verifier changed since collection') actual = getattr(series, 'verify_' + name)(directory) require({k: v for k, v in recorded.items() if k != 'verifier_sha256'} == actual, 'series raw result/reference drift') verified.update(actual['verified_conditions']) bindings[name] = binding['sha256'] return {'verified_conditions': sorted(verified), 'series_manifest_sha256': sha(manifest_path), 'C_source_manifest_sha256': sha(safe(c_base, 'SOURCE_MANIFEST.json')), 'C_provenance_sha256': sha(safe(c_base, 'OWNERSHIP_RAW_PROVENANCE.json')), 'series_bindings': bindings, 'relocation_manifest_sha256': series.MANIFEST_SHA256, 'missing_conditions': sorted((set(REQUIRED) | {'EV'}) - verified), 'nativeWindows': {name: 'pending' for name in sorted(set(REQUIRED) | {'EV'})}, 'stage_A': 'candidate binding stage A pending', 'formal_acceptance': False} def combine_native_series(directory, native): """Fresh raw supplies conditions; historical collection is separate provenance.""" import importlib.util def module(name, path): spec = importlib.util.spec_from_file_location(name, path) value = importlib.util.module_from_spec(spec) spec.loader.exec_module(value) return value module_path = ROOT / DOC_SCOPE / 'p04_native_series.py' series = module('p04_fresh_series', module_path) root_bindings = series.fresh_bindings(directory) collection = root_bindings['collection'] require(re.fullmatch(r'audit-support/p04-raw-collect-q[0-9]+', collection) is not None, 'exclusive fresh collection name') base = directory / collection manifest_path = safe(base, 'MANIFEST.json') manifest = load(manifest_path) require(manifest['protocol'] == 'p04-fresh-native-series-collection-v1' and manifest['historical_merged'] is False and set(manifest['series']) == set(series.FRESH_RUNS), 'fresh series roster') verified, bindings, results, pids = set(), {}, {}, [] for name in series.FRESH_RUNS: binding = manifest['series'][name] path = safe(base, binding['path']) require(path.name == name + '.json' and sha(path) == binding['sha256'] and path.stat().st_size == binding['bytes'], 'fresh series collection pin') recorded = load(path) require(recorded['verifier_sha256'] == sha(module_path), 'fresh verifier drift') actual = getattr(series, 'verify_fresh_' + name)(directory) require({k: v for k, v in recorded.items() if k != 'verifier_sha256'} == actual, 'fresh raw recomputation drift') verified.update(actual['verified_conditions']) pids.extend(actual['owned_pids']) bindings[name], results[name] = binding['sha256'], actual require(len(pids) == len(set(pids)), 'fresh series PID reuse') order = results['s5']['facts']['order']['cases'] gates = { 'D05-P04-DESTRUCTION': [('S5C_DESTRUCTION_COMMAND', 'command_destruction_before_completion'), ('S5D_DESTRUCTION_RAID', 'G1'), ('S5D_IN_FLIGHT', 'G2'), ('S5D_DESTRUCTION_RAID', 'idle_objective_breach'), ('S5D_IN_FLIGHT', 'in_flight_destruction'), ('S5D_PAID_BEFORE_INCIDENT', 'paid_before_incident')], 'D05-P04-MIDNIGHT-DAY': [('S5E_MIDNIGHT', 'midnight_close_upkeep_before_sale'), ('S5E_MIDNIGHT', 'midnight_new_day_sale')], 'D05-P04-EXIT': [('S5F_EXIT', 'command_exit_before_start'), ('S5F_EXIT', 'planned_exit_after_completion')], } for ref, pairs in gates.items(): require(all(order[case][key] is True for case, key in pairs), 'S5 order gate: ' + ref) for case, key in [('S5G_SHIFT_1080', 'shift_completion'), ('S5H_PAUSED_MIDNIGHT', 'paused_midnight'), ('S5E_MIDNIGHT', 'same_minute_replay')]: require(order[case][key] == REQUIRED[key][2], 'fresh S5 threshold: ' + key) power = results['s5']['facts']['power']['cases'] require(power['S5A_PURE']['measurements']['power_stop'] == REQUIRED['power_stop'][2] and power['S5B_DEPENDENCY']['measurements']['dependency_stop'] == REQUIRED['dependency_stop'][2], 'S5 D05-P04-PAUSE-REASON gate') full_verifier = module('p04_full_verifier', ROOT / DOC_SCOPE / 'full_suite_verifier.py') full_binding = root_bindings['full_suite'] full_run = directory / full_binding['run'] require(not Path(full_binding['run']).is_absolute() and '..' not in Path(full_binding['run']).parts, 'full run path') for leaf, key in [('runner-input.json', 'runner_input'), ('runner-ended.json', 'runner_ended')]: series.checked(safe(full_run, leaf).read_bytes(), full_binding[key], 'Root full receipt') full_inputs = load(safe(full_run, 'runner-input.json')) full = full_verifier.verify(Path(full_inputs['project_root']), full_run, REQUIRED) require(full['verified'] is True and set(full['conditions']) == set(REQUIRED) | {'EV'}, 'full condition membership') expected_source = {n: p['sha256'] for n, p in load(directory / root_bindings['series']['s5']['run'] / 'source_before.json').items()} require(full['source_sha256'] == expected_source, 'full and boundary exact same Q') history = safe(directory / 'audit-support/p04-raw-collect-r22', 'MANIFEST.json') c_base = directory / 'audit-support/native-first-current' require(native == verify_native(c_base), 'historical C provenance recomputation') return {'verified_conditions': sorted(verified), 'series_manifest_sha256': sha(manifest_path), 'series_bindings': bindings, 'fresh_collection': collection, 'full_suite': full, 'historical_collection': {'manifest_sha256': sha(history), 'merged': False}, 'C_source_manifest_sha256': sha(safe(c_base, 'SOURCE_MANIFEST.json')), 'C_provenance_sha256': sha(safe(c_base, 'OWNERSHIP_RAW_PROVENANCE.json')), 'missing_conditions': sorted((set(REQUIRED) | {'EV'}) - verified), 'nativeWindows': {name: 'pending' for name in sorted(set(REQUIRED) | {'EV'})}, 'stage_A': 'candidate binding stage A pending', 'formal_acceptance': False} def verify(directory): # Contract hashes the supplied spec string, while the checked-in JSON has # one final LF. Accept only that exact representation difference. spec_bytes = SPEC.read_bytes() require(sha(SPEC) == SPEC_HASH or (spec_bytes.endswith(b'\n') and hashlib.sha256(spec_bytes[:-1]).hexdigest() == SPEC_HASH), 'dispatch spec pin mismatch') spec = load(SPEC) evidence = Path(os.environ.get('P04_EVIDENCE_DIR', str(directory / 'p04-evidence'))).absolute() require(not evidence.is_symlink(), 'symlink evidence directory') index = load(safe(evidence, 'index.json')) require(type(index.get('schema_version')) is int and index['schema_version'] in (1, 2), 'schema_version') require(index.get('task_id') == 'P04-01' and index.get('formal_acceptance') is False, 'task/acceptance') head, target_result = verify_roles(index, directory) if index['schema_version'] == 2: import importlib.util module_path = ROOT / DOC_SCOPE / 'build_p04_index.py' module_spec = importlib.util.spec_from_file_location('p04_index', module_path) builder = importlib.util.module_from_spec(module_spec) module_spec.loader.exec_module(builder) expected = builder.payload(directory, sys.modules[__name__]) require(index == expected, 'forged index/raw mapping differs from recomputation') require(not index['raw_conditions']['missing_conditions'], 'uncollected original boundaries: ' + ', '.join(index['raw_conditions']['missing_conditions'])) return target_result source = index.get('source_map') applicability(index, spec, source, head) oldmap = load(ROOT / 'docs/plans/original-completion-2026-10-02/execution/p04-01-input-r01/historical-source-map.json') required = {p for p in oldmap if p.endswith('.gd') and p.startswith(('autoload/', 'scripts/'))} require(required <= source.keys(), 'missing runtime/service dependency source pins') environment = index.get('environment') require(isinstance(environment, dict) and isinstance(environment.get('os'), str) and environment['os'], 'environment OS') for kind in ('engine', 'configuration', 'inputs'): pins(environment.get(kind), evidence) artifacts = index.get('artifacts') require(isinstance(artifacts, dict) and artifacts, 'missing artifacts') paths, raw = set(), {} for aid, item in artifacts.items(): require(isinstance(aid, str) and aid and isinstance(item, dict), 'invalid artifact') path = safe(evidence, item['path']) require(path not in paths, 'duplicate artifact path') paths.add(path) require(type(item.get('bytes')) is int and item['bytes'] == path.stat().st_size, 'artifact size: ' + aid) require(sha(path) == item.get('sha256'), 'artifact hash: ' + aid) require(item.get('type') in {'observation', 'state', 'receipt', 'stdout', 'native_log', 'save', 'authority'}, 'artifact type') raw[aid] = path authority = index.get('authority') require(isinstance(authority, dict) and authority, 'missing authority') domains = index.get('domains') require(isinstance(domains, dict) and set(domains) == DOMAINS, 'required seven domains') for name, domain in domains.items(): require(isinstance(domain, dict), 'invalid domain: ' + name) for field in ('source_refs', 'authority_refs', 'cases', 'invariants'): require(isinstance(domain.get(field), list) and domain[field], name + ': missing ' + field) require(domain.get('applicability') == {'source_map': source, 'environment': environment}, name + ': applicability mismatch') for ref in domain['source_refs']: require(isinstance(ref, dict) and ref.get('path') in source and isinstance(ref.get('function'), str) and ref['function'], 'invalid source ref') require(ref.get('sha256') == source[ref['path']], 'source ref hash') for ref in domain['authority_refs']: require(ref in authority and authority[ref].get('status') in {'original', 'approved'}, 'pending/missing authority') require(authority[ref].get('artifact_id') in raw, 'missing raw authority') case_ids = set() for case in domain['cases']: require(isinstance(case, dict) and isinstance(case.get('id'), str) and case['id'] not in case_ids, 'duplicate/invalid case') case_ids.add(case['id']) for field in ('input', 'before', 'operation', 'after', 'receipt'): require(case.get(field) in raw, name + ': missing case artifact ' + field) receipt = load(raw[case['receipt']]) pid = receipt.get('start_pid') require(type(pid) is int and pid > 0 and receipt.get('end_pid') == pid and type(receipt.get('exit_code')) is int and receipt['exit_code'] == 0 and receipt.get('engine_ended') is True, 'unexecuted/incomplete receipt') require(receipt.get('source_map') == source and receipt.get('environment') == environment, 'receipt applicability') for field in ('stdout', 'native_log'): require(receipt.get(field) in raw, 'receipt missing raw ' + field) require(receipt.get('case_id') == case['id'], 'receipt case identity') require(receipt.get('artifact_hashes') == {field: sha(raw[case[field]]) for field in ('input', 'before', 'operation', 'after')}, 'receipt artifact binding') for pred in domain['invariants']: require(isinstance(pred, dict) and pred.get('artifact_id') in raw, 'missing predicate artifact') require(pred.get('authority_ref') in domain['authority_refs'], 'predicate authority') require(pred.get('raw_origin') == 'engine', 'unsupported derivation/origin') value = pointer(load(raw[pred['artifact_id']]), pred['json_pointer']) op = pred.get('operator') if op in {'same_literal', 'distinct'}: require(pred.get('paired_artifact_id') in raw, 'missing paired artifact') other = pointer(load(raw[pred['paired_artifact_id']]), pred['paired_json_pointer']) result = literal(value, other) if op == 'same_literal' else not literal(value, other) elif op == 'eq': result = literal(value, pred['expected']) elif op == 'nonnegative': result = type(value) in (int, float) and value >= 0 elif op == 'count_eq': result = isinstance(value, (list, dict)) and type(pred.get('expected')) is int and len(value) == pred['expected'] else: raise ValueError('unsupported predicate: ' + str(op)) require(result, name + ': predicate failed') require(index.get('review_status') in {'pending_review', 'approved'}, 'review status') coverage(index, raw, source, head) # Scalar conditions remain necessary declarations, never sufficient proof. binding = index.get('native_bundle') require(isinstance(binding, dict), 'native bundle required; self-reported scalar evidence is insufficient') manifest = supplied_file(directory, binding) require(manifest.name == 'OWNERSHIP_RAW_PROVENANCE.json' and sha(manifest) == binding['sha256'], 'native bundle provenance pin') native = verify_native(manifest.parent) require({safe(manifest.parent, run + '/runner-input.json') for run in NATIVE_RUNS} <= target_result['runner_paths'], 'C runner inputs absent from complete product binding') require(native['production_commit'] == target_result['tested_product'], 'native production role mismatch') # This adapter covers only the supplied dependency/formal series. Never # promote it to the missing original scenarios through scalar booleans. combined = combine_native_series(directory, native) missing = set(combined['missing_conditions']) require(not missing, 'uncollected original boundaries: ' + ', '.join(sorted(missing)) + '; formal_original_credit=0') return target_result def main(): parser = argparse.ArgumentParser(description=__doc__) mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument('--verify-only', action='store_true') mode.add_argument('--verify-series-only', action='store_true', help='Read-only Stage R composition; always leaves Stage A pending') args = parser.parse_args() try: require(bool(os.environ.get('BUILD_OUTPUT_DIR')), 'BUILD_OUTPUT_DIR required') if args.verify_series_only: directory = Path(os.environ['BUILD_OUTPUT_DIR']).absolute() native = verify_native(directory / 'audit-support/native-first-current') result = combine_native_series(directory, native) result.update(task_id='P04-01', machine_verified=False, error='candidate binding stage A pending; uncollected original boundaries: ' + ', '.join(result['missing_conditions'])) print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 1 with isolated_objects(): result = verify(Path(os.environ['BUILD_OUTPUT_DIR']).resolve()) except (ValueError, OSError, KeyError, TypeError, IndexError, AttributeError, subprocess.SubprocessError) as exc: print(json.dumps({'task_id': 'P04-01', 'machine_verified': False, 'formal_acceptance': False, 'error': str(exc)}, ensure_ascii=False)) return 1 print(json.dumps({'task_id': 'P04-01', 'machine_verified': True, 'formal_acceptance': False, 'role': result['role'], 'actual_head': result['actual_head']})) return 0 if __name__ == '__main__': sys.exit(main())