P18-01-CODE: 原本関係モデル・旧値保全・Runtime保存と読取経路

This commit is contained in:
Agent Control Runner
2026-10-13 14:40:02 +09:00
parent f76ad96990
commit 0084760ee0
2 changed files with 544 additions and 48 deletions
@@ -0,0 +1,80 @@
# P18-01-CODE recorded repair 2/2: collection / verification split
Resumes candidate `f76ad96990a4d7371b6ab667df0a068cc06be798`. No counter reset,
game/test reconstruction, Runner change, commit or integration. Changes are confined
to the narrow collector and this note. Existing game behavior evidence remains the
retained Astra preflight; this worker did not rerun or reinterpret it as final proof.
## Modes
- Default: `python -B tools/run_p18_relationship_tests.py --output {build_dir}/p18-relationship-evidence`.
Read-only verification in the ordinary registered sandbox. No subprocess call,
process discovery, native launch, or access to the external player/private profiles.
Current checkout source/settings/support/marker bytes and the locally configured
`godot` binary are read for pins. Receipt-provided paths never select an external
file to open. Original absolute paths remain historical identity strings.
- `--collect`: Root orchestration only, outside Seatbelt. Uses the existing
`source_refs`, private source copy, `execute_phase`, owned_outer, Guard, logger,
diagnostic, marker and bounded cleanup contracts unchanged. Fresh output outside
the checkout, with an existing parent, is required. Runs import, all 308 scripts,
and exactly the seven fixed tests once. Owned profile bytes are archived after
each phase. The final verifier runs after collection; no duplicate regression.
“Root” identifies the authorized orchestrator, not a requirement to run as Unix uid 0.
- `--prepare-only`: reports source bytes, available capacity, 308 parse paths and
the exact seven-test/nine-phase roster; creates no stage/profile and starts no engine.
Only `--output`, `--collect`, `--prepare-only` and help are accepted. Duplicate,
abbreviated, unknown and malformed flags, combined modes, traversal, symlinks,
nonregular evidence and hardlinked evidence reject. Verification never rewrites
raw paths or files. The ordinary-owner pure proof reader maps only the original
output prefix to the supplied output and checks confinement before opening files.
## Receipt schema
`receipt.json` (`guarded-p18-v2`) is written last after successful collection.
It contains the full relative source content map (bytes/SHA-256, no checkout-path
dependency), its canonical JSON SHA-256, project and marker-contract SHA-256,
engine SHA-256, and a complete relative file inventory with byte sizes/SHA-256.
The inventory covers the retained private source, engine bytes, bootstraps,
starters, raw stdout/stderr/native logger, phase receipts, START/END/process and
both cleanup records, Guard reports, ordinary/audio ownership proofs, typed
snapshots, and complete archived startup/runtime profiles. Only the enclosing
receipt itself is outside its own inventory.
Verification checks the inventory and recomputes raw marker/diagnostic gates,
source and helper pins, exact commands, START/END identity, all nine distinct PIDs
and bounded cleanup, Guard conditions, proof gates, complete player before/after
inventory equality, and seven test rows against nine phase receipts. It requires
one exact `FULL_DEPENDENCY_PARSE checks=308 failures=0` marker. Old v1, absent,
incomplete, mismatched or altered evidence rejects. A source change, including
this collector, requires fresh Root collection; no source exemptions exist.
## Worker checks and environment observations
AST/in-memory compile and prepare-only passed. Prepare-only found 1311 source
files, 308 parse scripts, seven tests and nine future phases, with sufficient
reported capacity. Synthetic negative checks rejected missing receipts, changed
hashes, symlinks, relative and absolute escapes, incomplete files and invalid or
duplicate flags/mode combinations. Those checks replaced subprocess entry points
and default-player access with fail-fast functions; neither was called. No native
test, full regression, `--collect`, Runner operation or review was executed.
Environment evidence: a shell here-document failed with
`zsh:1: can't create temp file for here document: operation not permitted`.
The `python` alias was unavailable (`zsh:1: command not found: python`);
worker checks used `/usr/bin/python3 -B`. Ordinary authorized file edits and
temporary negative fixtures under the task allocation succeeded. No isolation
or permission policy was changed. Root/Runner must provide the configured
`python` command used by the fixed validator argv.
## Root next steps
1. On the final unchanged source, run the native `--collect` exactly once outside
Seatbelt into the allocation build directory's fresh `p18-relationship-evidence`.
Do not reuse old v1 evidence or run collection from ordinary Runner validation.
2. Run ordinary Runner validation using the fixed default verify-only argv.
3. Obtain a separate Opus review, then perform the authorized integration and
affected final validation. This worker submits implementation, not acceptance.
P18-01 remains incomplete: ordinary UI rendering/display equality, approved initial
values, the full final roster and applicable Windows final acceptance remain unmet.
+464 -48
View File
@@ -1,9 +1,11 @@
#!/usr/bin/env python3
"""Runner-owned narrow P18 collection using the unchanged guarded phase protocol."""
"""Root-owned narrow P18 collection using the unchanged guarded phase protocol."""
import argparse
import hashlib
import json
import os
import re
import stat
from pathlib import Path
import shutil
import sys
@@ -23,55 +25,14 @@ ROSTER = [
]
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--project-root', type=Path, default=Path(__file__).resolve().parent.parent)
parser.add_argument('--output', type=Path)
parser.add_argument('--godot', default='godot')
parser.add_argument('--timeout', type=int, default=1200)
parser.add_argument('--prepare-only', action='store_true')
args = parser.parse_args(argv)
root = args.project_root.resolve()
out = (args.output or root / 'artifacts' / ('p18-relationship-' + uuid.uuid4().hex)).resolve()
shared.require(args.timeout > 0, 'positive timeout')
refs = shared.source_refs(root)
# Refuse mixed collector/support checkouts; all actual Python source is pinned.
for name in ['tools/run_all_tests.py', 'tools/run_p18_relationship_tests.py', 'tools/strict_test_markers.py']:
shared.require(refs[name]['sha256'] == shared.sha(shared.HERE / Path(name).name), 'actual collector source ' + name)
for file in shared.SUPPORT.iterdir():
if file.is_file():
shared.require(refs['tools/regression/' + file.name]['sha256'] == shared.sha(file), 'actual support source ' + file.name)
shared.require(all(name in refs for name in ROSTER), 'complete narrow roster')
shared.require(set(ROSTER).issubset(shared.roster(root)), 'normal discovery includes narrow roster')
markers = shared.module('p18_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 ROSTER[:2]:
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')]
def collect(args, root, out, refs, markers, contracts, policy, contract_path, contract_sha, parse_paths):
source_bytes = sum(pin['bytes'] for pin in refs.values())
ancestor = out.parent
while not ancestor.exists():
ancestor = ancestor.parent
free = shutil.disk_usage(ancestor).free
free = shutil.disk_usage(out.parent).free
needed = source_bytes + shared.MIN_SPACE
preview = dict(roster=ROSTER, source_files=len(refs), source_bytes=source_bytes,
free_bytes=free, conservative_required_bytes=needed, capacity_available=free >= needed,
parse_paths=len(parse_paths), future_native_phases=len(ROSTER) + 2,
actual_checks=None, engines=0, stages=0, formal_acceptance=False)
if args.prepare_only:
print(json.dumps(preview, ensure_ascii=False))
return 0
shared.require(free >= needed, 'fresh source and phase disk admission')
outer = shared.module('p18_owned_outer', shared.SUPPORT / 'owned_outer.py')
shared.require(not outer.native.helper['live_engines'](), 'another Godot owns CPU')
engine = shutil.which(args.godot) if not Path(args.godot).is_absolute() else args.godot
engine = shutil.which('godot')
shared.require(engine and Path(engine).is_file(), 'native launcher exists')
engine = str(Path(engine).resolve())
out.mkdir(parents=True, exist_ok=False)
@@ -94,7 +55,7 @@ def main(argv=None):
public = (stage / 'project.godot').read_bytes()
shared.write(out / 'parse-paths.json', parse_paths)
run_id = uuid.uuid4().hex
shared.write(out / 'runner-input.json', dict(protocol='guarded-p18-v1', run_id=run_id,
shared.write(out / 'runner-input.json', dict(protocol='guarded-p18-v2', run_id=run_id,
owner_runner_pid=os.getpid(), project_root=str(root), stage=str(stage), roster=ROSTER,
selected_source_map=selected['map'], marker_contract_sha256=contract_sha,
runner_sha256=shared.sha(Path(__file__)), shared_runner_sha256=shared.sha(shared.HERE / 'run_all_tests.py'),
@@ -112,6 +73,7 @@ def main(argv=None):
shared.require(shared.sha(contract_path) == contract_sha, 'marker contract unchanged')
row = shared.execute_phase(engine, stage, out, phase, test, selected, expected, previous,
policy, contracts, parse_paths, public, protected, root, outer, markers, diagnostics, owners, audio)
archive_profiles(out, row)
phases.append(row)
if test:
row.update(sequence=len(rows), godot=engine)
@@ -134,16 +96,470 @@ def main(argv=None):
ended = bool(outer.OWNED) and all(row['engine_ended'] is True for row in cleanup)
passed = error is None and not cleanup_errors and not guard_errors and ended and len(phases) == len(ROSTER) + 2 and all(row['passed'] for row in phases)
context = shared.totals(rows, ROSTER)
shared.write(out / 'runner-ended.json', dict(context, protocol='guarded-p18-v1', run_id=run_id,
shared.write(out / 'runner-ended.json', dict(context, protocol='guarded-p18-v2', run_id=run_id,
owner_runner_pid=os.getpid(), passed=passed, engine_ended=ended,
actual_captured_PIDs=[process.pid for process in outer.OWNED], final_owned_cleanup=cleanup,
cleanup_errors=cleanup_errors, guard_errors=guard_errors, error=error, phases=phases,
all_tests_sha256=shared.sha(out / 'all-tests.json') if (out / 'all-tests.json').exists() else None,
default_player_after=shared.default_player(root), formal_acceptance=False,
original_unit_complete=False, independent_review='pending', native_Windows='pending'))
if passed:
seal(out, refs, engine, contract_sha)
# Root also exercises precisely the verifier that ordinary Runner uses.
verify(out, refs, markers, contracts, policy, contract_sha, parse_paths)
print(json.dumps(dict(context, passed=passed, error=error, cleanup_errors=cleanup_errors, guard_errors=guard_errors)), flush=True)
return 0 if passed else 1
# Receipt file names are always relative; absolute paths in native receipts are
# historical identity strings only and are never opened by the verifier.
def require(ok, reason):
shared.require(ok, reason)
def digest(value):
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':'),
ensure_ascii=False).encode()).hexdigest()
def content_map(refs):
return {name: {'bytes': pin['bytes'], 'sha256': pin['sha256']}
for name, pin in refs.items()}
def plain_path(path):
path = Path(path)
require('..' not in path.parts, 'parent traversal refused')
absolute = Path(os.path.abspath(path))
for part in [absolute, *absolute.parents]:
require(not part.is_symlink(), 'symlink path refused: ' + str(part))
return absolute
def confined(out, name):
require(type(name) is str and name and '\\' not in name, 'ordinary relative evidence name')
path = Path(name)
require(not path.is_absolute() and '..' not in path.parts
and path.as_posix() == name and name != '.', 'evidence path escape or alias')
result = plain_path(out / path)
require(result.is_relative_to(out), 'evidence confined to output')
return result
def inventory(out):
result = {}
require(out.is_dir(), 'output directory required')
for directory, dirs, files in os.walk(out, followlinks=False):
for name in dirs + files:
p = Path(directory) / name
mode = p.lstat().st_mode
require(stat.S_ISREG(mode) or stat.S_ISDIR(mode), 'nonordinary evidence refused')
if stat.S_ISREG(mode):
require(p.stat().st_nlink == 1, 'hardlinked evidence refused')
result[p.relative_to(out).as_posix()] = {
'bytes': p.stat().st_size, 'sha256': shared.sha(p)}
return result
def unique_object(pairs):
result = {}
for key, value in pairs:
require(key not in result, 'duplicate JSON key')
result[key] = value
return result
def read_json(path):
return json.loads(path.read_text(encoding='utf-8'), object_pairs_hook=unique_object,
parse_constant=lambda token: require(False, 'nonfinite JSON ' + token))
def archive_profiles(out, row):
for label in ('startup', 'runtime'):
value = row.get('whole_' + label + '_profile_after')
if value is None:
continue
dest = out / 'profiles' / row['phase'] / label
dest.mkdir(parents=True, exist_ok=False)
source = plain_path(value['path'])
require(shared.inventory(source) == value['files'], 'whole owned profile before archive')
for name in value['files']:
target = confined(out, str((dest / name).relative_to(out)))
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(source / name, target)
require(shared.inventory(dest) == value['files'] == shared.inventory(source),
'whole owned profile archive exact')
def seal(out, refs, engine, contract_sha):
# Last file written: an interrupted collection has no complete receipt.
shutil.copyfile(engine, out / 'engine.bin')
require(shared.sha(out / 'engine.bin') == shared.sha(engine), 'engine archive exact')
source = content_map(refs)
shared.write(out / 'receipt.json', {
'schema': 'guarded-p18-v2', 'source_map': source,
'source_map_sha256': digest(source),
'project_settings_sha256': source['project.godot']['sha256'],
'marker_contract_sha256': contract_sha,
'engine_sha256': shared.sha(engine),
'files': inventory(out),
})
def cleanup_gate(value):
require(value['owned_child_created'] is True and value['engine_ended'] is True
and value['actual_process_exit_code'] == 0 and value['errors'] == [],
'successful actual owned cleanup')
attempts = value['attempts']
require(1 <= len(attempts) <= 3, 'bounded cleanup attempts')
for index, row in enumerate(attempts):
require(row['attempt'] == index + 1 and row['errors'] == []
and row['wait_timeout_seconds'] == 5 and row['wait_completed'] is True
and row['wait_returned_code'] == 0 and row['poll_after'] == 0,
'complete bounded owned wait')
def outer_cleanup(rows, pids):
require([r['pid'] for r in rows] == pids, 'exact captured cleanup PID order')
for row in rows:
require(row['engine_ended'] is True and row['outer_cleanup_errors'] == []
and row['foreign_process_or_group_signals'] == 0
and row['outer_maximum_helper_invocations'] == 3
and row['helper_maximum_waits'] == 3 and row['per_wait_timeout_seconds'] == 5,
'unchanged outer cleanup contract')
cleanup_gate(row['cleanup'])
def verify(out, refs, markers, contracts, policy, contract_sha, parse_paths):
receipt_path = confined(out, 'receipt.json')
require(receipt_path.is_file(), 'regular final receipt required')
manifest = read_json(receipt_path)
require(manifest['schema'] == 'guarded-p18-v2', 'current receipt schema required')
files = manifest['files']
require(type(files) is dict and files, 'complete raw manifest')
for name, pin in files.items():
p = confined(out, name)
require(p.is_file() and set(pin) == {'bytes', 'sha256'}
and type(pin['bytes']) is int and pin['bytes'] >= 0
and re.fullmatch('[0-9a-f]{64}', pin['sha256']) is not None,
'ordinary artifact pin')
actual = inventory(out)
receipt_pin = actual.pop('receipt.json')
require(actual == files, 'complete raw file set and SHA-256 equality')
source = content_map(refs)
require(manifest['source_map'] == source and manifest['source_map_sha256'] == digest(source),
'current whole source content map independent of checkout path')
require(manifest['project_settings_sha256'] == source['project.godot']['sha256']
and manifest['marker_contract_sha256'] == contract_sha, 'current project/marker pins')
def read(name):
require(name in files, 'required raw file missing: ' + name)
return read_json(confined(out, name))
def text(name):
require(name in files, 'required raw transport missing: ' + name)
return confined(out, name).read_text(encoding='utf-8')
start, end = read('runner-input.json'), read('runner-ended.json')
require(start['protocol'] == end['protocol'] == 'guarded-p18-v2'
and start['run_id'] == end['run_id']
and re.fullmatch('[0-9a-f]{32}', start['run_id']) is not None,
'same current run')
require(type(start['owner_runner_pid']) is int and start['owner_runner_pid'] > 0
and start['owner_runner_pid'] == end['owner_runner_pid'], 'same collector owner')
require(start['roster'] == ROSTER and len(set(ROSTER)) == 7, 'exact seven unique tests')
require(start['selected_source_map'] == {n: p['sha256'] for n, p in source.items()}
and content_map(read('source-start.json')) == source, 'whole selected source pins')
for field, name in [('runner_sha256', 'tools/run_p18_relationship_tests.py'),
('shared_runner_sha256', 'tools/run_all_tests.py'),
('marker_module_sha256', 'tools/strict_test_markers.py')]:
require(start[field] == source[name]['sha256'], 'current collector/shared/marker pin')
require(start['marker_contract_sha256'] == contract_sha, 'marker contract pin')
# Resolve only the locally configured launcher, never the receipt's engine path.
engine = shutil.which('godot')
require(engine and shared.sha(engine) == start['engine_sha256']
== manifest['engine_sha256'] == files['engine.bin']['sha256'],
'current native engine and retained bytes pinned')
require(read('parse-paths.json') == parse_paths and len(parse_paths) == 308,
'all 308 exact source parse paths')
require(end['passed'] is True and end['engine_ended'] is True and end['error'] is None
and end['cleanup_errors'] == [] and end['guard_errors'] == [],
'finished successful collection without guard/cleanup error')
for receipt in (start, end):
require(receipt['formal_acceptance'] is False and receipt['original_unit_complete'] is False,
'backend evidence never completes original unit')
before = start['default_player_before']
require(before == end['default_player_after'] and before['logs_and_locks_included'] is True
and type(before['existed']) is bool and type(before['files']) is dict,
'complete default player before/after equality')
for name, pin in before['files'].items():
confined(out, name) # Validate the inventory name; never open the player.
require(set(pin) == {'bytes', 'sha256'} and type(pin['bytes']) is int
and pin['bytes'] >= 0 and re.fullmatch('[0-9a-f]{64}', pin['sha256']),
'whole default player file pin')
require(before['existed'] or before['files'] == {}, 'absent player has no files')
native = shared.module('p18_verify_native', shared.SUPPORT / 'owned_diagnostic_process.py')
diagnostics = shared.module('p18_verify_diagnostics', shared.SUPPORT / 'diagnostic_gate.py')
owners = shared.module('p18_verify_owners', shared.SUPPORT / 'ordinary_proof_gate.py')
# Helpers are loaded for their pure gates only. No run/probe/cleanup is called.
origin = Path(start['stage']).parent
require(Path(start['stage']) == origin / 'stage' and origin.is_absolute()
and '..' not in origin.parts, 'original output identity')
def relocated(value):
if isinstance(value, dict):
return {relocated(k): relocated(v) for k, v in value.items()}
if isinstance(value, list):
return [relocated(v) for v in value]
if isinstance(value, str) and value.startswith(str(origin) + '/'):
return str(confined(out, value[len(str(origin)) + 1:]))
return value
def proof_read(path):
path = plain_path(path)
require(path.is_relative_to(out), 'proof path stays in output')
return relocated(read(path.relative_to(out).as_posix()))
owners.read = proof_read
phases = end['phases']
require(len(phases) == 9, 'exact nine phases')
tests = read('all-tests.json')
require(len(tests) == 7 and end['all_tests_sha256'] == files['all-tests.json']['sha256'],
'complete test rows')
pids = []
for index, (phase, test) in enumerate([('import', None), ('parse', None)]
+ [(str(i).zfill(3), n) for i, n in enumerate(ROSTER)]):
row = read(phase + '-receipt.json')
expected_row = dict(row)
if test:
expected_row.update(sequence=index - 2, godot=start['engine'])
require(tests[index - 2] == expected_row, 'test row matches raw phase receipt')
require(phases[index] == expected_row and row['phase'] == phase and row['test'] == test
and row['passed'] is True and row['failed'] is False
and row['validation_error'] is None and not row.get('guard_error')
and row['default_player_entire_profile_unchanged'] is True,
'exact successful phase')
pid = row['actual_pid']
require(type(pid) is int and pid > 0 and pid not in pids, 'nine distinct actual PIDs')
pids.append(pid)
process = read(phase + '/process.json')
require(process == row['native_process'] and process['pid'] == pid
and process['owned_child_process_helper_sha256'] == native.HELPER_SHA
== source['tools/regression/p09_owned_engine_process.py']['sha256'],
'actual phase/process/helper agreement')
stdout, stderr, logger = (text(phase + '/' + n) for n in
('stdout.log', 'stderr.log', 'engine-native.log'))
raw = stdout + stderr
require(text(row['log']) == raw and row['log_sha256'] == files[row['log']]['sha256'],
'literal complete raw stream')
require(process['stdout_sha256'] == files[phase + '/stdout.log']['sha256']
and process['stderr_sha256'] == files[phase + '/stderr.log']['sha256'],
'process transport SHA')
require(row['raw_native_log'] == dict(files[phase + '/engine-native.log'],
path=str(origin / phase / 'engine-native.log')), 'native logger pin')
review = diagnostics.review_streams(native, {
'stdout': stdout, 'stderr': stderr, 'native_file': logger}, test)
require(row['finite_original_diagnostic_review'] == review
and diagnostics.diagnostic_result_gate(process, review)
and shared.process_gate(process, review['allowed_e31_error_lines']),
'strict diagnostics/process gates')
require(native.native_diagnostics(stdout, stderr, row['command'])
== process['strict_native_diagnostics'], 'raw banner/diagnostics recomputed')
command = [start['engine'], '--headless', '--path', start['stage']]
if phase == 'import':
command += ['--editor', '--import', '--quit', '--verbose']
else:
script = ('bootstrap/parse_full_dependencies.gd' if phase == 'parse'
else 'starters/' + phase + '.gd')
command += ['--script', str(origin / script), '--verbose']
require(row['starter_sha256'] == files[script]['sha256'], 'exact starter pin')
if test:
expected_starter = ('extends ' + json.dumps('res://' + test) + '\nconst Guard := preload('
+ json.dumps(str(origin / 'bootstrap/profile_guard.gd'))
+ ')\nfunc _initialize() -> void:\n\tif not Guard.activate():\n\t\tquit(97)\n\t\treturn\n\tsuper._initialize()\n')
require(text(script) == expected_starter, 'unchanged guarded starter bytes')
command += ['--log-file', str(origin / phase / 'engine-native.log')]
require(row['command'] == process['argv'] == command, 'exact native argv/logger')
for filename, field, event in [('started.json', 'actual_start_binding', 'ENGINE_STARTED'),
('ended.json', 'actual_end_binding', 'ENGINE_ENDED')]:
binding = read(phase + '/' + filename)
require(binding == process[field] and binding['event'] == event
and binding['actual_pid'] == binding['pid'] == pid
and binding['owner_pid'] == start['owner_runner_pid']
and binding['command'] == binding['argv'] == command, 'actual START/END ownership')
binding = process['actual_start_binding']
identity = binding['current_process_identity']
if identity is None:
require(binding['tracked_child_ended_before_initial_probe'] is True, 'early ended tracked child')
else:
require(binding['live_identity_exact'] is True and identity == {
'pid': pid, 'ppid': start['owner_runner_pid'], 'pgid': pid, 'args': ' '.join(command)},
'recorded live child PID/PPID/PGID/argv')
finish = process['actual_end_binding']
require(finish['engine_ended'] is True and finish['timed_out'] is False
and finish['lifecycle_error'] is None and finish['exit_code'] == 0
and finish['actual_process_exit_code'] == 0
and finish['start_receipt_written'] is True
and finish['monotonic_ns'] >= binding['monotonic_ns'], 'complete native END')
cleanup = read(phase + '/adapter-owned-cleanup.json')
require(cleanup == process['adapter_owned_cleanup'], 'raw adapter cleanup')
cleanup_gate(cleanup)
capture = row['outer_actual_capture']
require(read(phase + '/outer-owned-finally.json') == {
k: v for k, v in capture.items()
if k not in ('outer_receipt_written', 'outer_receipt_error', 'passed')},
'whole raw outer receipt')
require(capture['actual_captured_pids'] == [pid] and capture['passed'] is True
and capture['all_captured_ended'] is True and capture['adapter_exception'] is None
and capture['outer_cleanup_errors'] == [] and capture['outer_receipt_error'] is None
and capture['outer_receipt_written'] is True
and capture['all_cleanup_before_outer_receipt'] is True
and capture['foreign_process_or_group_signals'] == 0
and row['engine_ended'] is True and row['exit_code'] == 0, 'owned outer capture')
outer_cleanup(capture['cleanup'], [pid])
startup = read(phase + '-startup-profile-before.json')
require(startup == {'path': row['whole_startup_profile_after']['path'], 'existed': False,
'files': {}, 'before_native_spawn': True}, 'fresh startup profile')
require(row['whole_startup_profile_after']['fresh_before_engine'] is True,
'startup fresh before engine')
name = row['profile_startup_name']
require(re.fullmatch('TheTree-PCK-tests-[0-9a-f]{32}-standard-startup-' + phase, name)
and Path(startup['path']).name == name and row['private_project_fields'] == 2
and row['private_project_sha256'] == hashlib.sha256(shared.private_project(
(shared.HERE.parent / 'project.godot').read_bytes(), name)).hexdigest(),
'exact two-field startup isolation')
for label in ('startup', 'runtime'):
if label == 'runtime' and phase == 'import':
continue
whole = row['whole_' + label + '_profile_after']
archive = confined(out, 'profiles/' + phase + '/' + label)
require(archive.is_dir() and shared.inventory(archive)
== whole['files'], 'complete retained private profile')
if phase != 'import':
guard = read(phase + '-profile.json')
runtime = Path(row['whole_runtime_profile_after']['path'])
name = guard['custom_user_dir_name']
require(re.fullmatch('TheTree-PCK-tests-[0-9a-f]{32}-standard-runtime-' + phase, name)
and runtime.name == name and guard['verified'] is True
and guard['conditions'] == [True] * 5 and guard['conditions_checked'] == 5
and guard['fresh_before_parent_initialize'] is True
and guard['profile_writes_before_guard'] == 0
and guard['actual_user_data_dir'] == guard['expected_user_data_dir'] == str(runtime)
and guard['globalized_probe'] == str(runtime / 'guard-probe')
and guard['project_settings_persisted'] is False, 'all unchanged Guard conditions')
require(row['actual_private_profile'] == {
'report': guard, 'report_sha256': files[phase + '-profile.json']['sha256'],
'whole_files': row['whole_runtime_profile_after']['files']}, 'whole raw Guard report')
require(row['controlled_environment'] == {
'THE_TREE_PACK_PROFILE_NAME': name, 'THE_TREE_PACK_PROFILE_EXPECTED': str(runtime),
'THE_TREE_PACK_PROFILE_REPORT': str(origin / (phase + '-profile.json')),
'THE_TREE_FULL_PARSE_PATHS': str(origin / 'parse-paths.json'),
**({'THE_TREE_AUDIO_TEARDOWN_PROOF_DIR': str(origin / (phase + '-teardown/owned-audio')),
'THE_TREE_AUDIO_FIXTURE_CONTROL_PROOF': str(origin / (phase + '-teardown/control.json'))}
if test and policy['suite_policies'].get(test, {}).get('ordinary_cleanup_prepared') else {})},
'exact controlled Guard/proof environment')
else:
require(row['controlled_environment'] == {}, 'fresh import environment')
if phase == 'parse':
require(re.findall(r'(?m)^FULL_DEPENDENCY_PARSE checks=([0-9]+) failures=([0-9]+)\n', raw)
== [('308', '0')] and row['actual_parse_checks'] == 308, 'unique all-script parse marker')
if test:
parsed = markers.parse_raw(test, raw, contracts, source[test]['sha256'])
require(parsed['passed'] and row['completion'] == parsed
and row['reported_checks'] == parsed['checks']
and row['reported_failures'] == parsed['failures']
and row['completion_marker'] == parsed['marker'] and row['collector_errors'] == [],
'source-pinned raw completion marker')
require(row['source_sha256'] == row['source_sha256_end'] == source[test]['sha256'],
'exact test source before/after')
p = shared.phase_policy(test, Path(refs[test]['path']), policy['suite_policies'])
require(row['policy'] == p, 'current exact suite policy')
if p['ordinary_cleanup_prepared']:
proof = owners.ordinary_gate(confined(out, phase + '-teardown'), pid, out,
p['ordinary_control_labels'])
require(proof == relocated(row['ordinary_owner_pairs']), 'full ordinary/audio ownership proof')
require(row['generated_reports'] == {}, 'fixed narrow roster has no generated source reports')
require(end['actual_captured_PIDs'] == pids, 'all nine captured engines')
outer_cleanup(end['final_owned_cleanup'], pids)
require(all(end[k] == v for k, v in shared.totals(tests, ROSTER).items()), 'recomputed exact totals')
require(shared.inventory(confined(out, 'stage'), stage=True) == source, 'entire retained source/settings exact')
for name, pin in shared.read(shared.SUPPORT / 'bootstrap-scripts.json').items():
require(text('bootstrap/' + name).encode() == pin['text'].encode()
and files['bootstrap/' + name]['sha256'] == pin['sha256'], 'unchanged bootstrap/Guard pins')
require(inventory(out) == dict(files, **{'receipt.json': receipt_pin}),
'whole raw evidence unchanged during verification')
require(content_map(shared.source_refs(shared.HERE.parent)) == source,
'whole current source unchanged during verification')
print(json.dumps({'verified': True, 'actual_suites': 7, 'parse_checks': 308,
'actual_checks': end['actual_all_roster_checks'], 'engines_started': 0,
'formal_acceptance': False, 'original_unit_complete': False}))
return 0
def main(argv=None):
argv = list(sys.argv[1:] if argv is None else argv)
# No argparse abbreviation, repeated flags, = aliases or positional tokens.
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument('--output', required=True)
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 = set()
i = 0
while i < len(argv):
flag = argv[i]
require(flag in ('--output', '--collect', '--prepare-only', '--help')
and flag not in seen, 'unknown, duplicate or malformed flag')
seen.add(flag)
if flag == '--output':
i += 1
require(i < len(argv) and argv[i] and not argv[i].startswith('--'), 'output value required')
i += 1
args = parser.parse_args(argv)
args.timeout = 1200
root = shared.HERE.parent
out = plain_path(args.output)
refs = shared.source_refs(root)
require(set(ROSTER).issubset(shared.roster(root)), 'normal discovery includes narrow roster')
markers = shared.module('p18_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)
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 ROSTER[:2]:
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')]
require(len(parse_paths) == 308, 'exact expected parse count')
if args.prepare_only:
ancestor = out.parent
while not ancestor.exists():
ancestor = ancestor.parent
free = shutil.disk_usage(ancestor).free
size = sum(p['bytes'] for p in refs.values())
print(json.dumps({'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': 308,
'future_native_phases': 9, 'actual_checks': None, 'engines': 0, 'stages': 0,
'formal_acceptance': False}))
return 0
if args.collect:
require(not out.exists() and out.parent.is_dir(), 'fresh output with existing parent required')
require(not out.is_relative_to(root), 'private collection outside source tree')
return collect(args, root, out, refs, markers, contracts, policy,
contract_path, contract_sha, parse_paths)
return verify(out, refs, markers, contracts, policy, contract_sha, parse_paths)
if __name__ == '__main__':
raise SystemExit(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)