chickenbstrips/TheTree_mock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P08-03: 配置の通常操作受入

Browse Source
This commit is contained in:
Agent Control Runner
2026-10-14 03:22:38 +09:00
parent 12430ba899
commit 7dd450c64c
22 changed files with 7543 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/check_p08_normal_placement_evidence.py
+440
View File
@@ -0,0 +1,440 @@
#!/usr/bin/env python3
"""Read-only P08-03 evidence gate. Integrity success is never unit acceptance."""
import argparse
import hashlib
import importlib.util
import json
import math
import os
from pathlib import Path
import re
import sys
import struct
import zlib
import subprocess
sys.dont_write_bytecode = True
HERE = Path(__file__).resolve().parent
SPEC = 'docs/spec/acceptance/P08-03.json'
SPEC_SHA = '7941da4edbd3c43a636a4a3d834d57c095b56da988fa3a056e55b22d3a5d50e9'
CONTRACT = 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json'
TEST = 'tests/ui/test_p08_normal_placement_ops.gd'
RECORDER = 'docs/plans/original-completion-2026-10-02/execution/p08-03/passive_os_placement_recorder.gd'
CONDITIONS = ['support_connection', 'invalid_position_reason', 'demolish_rebuild',
'pan_zoom_world_identity', 'separate_pid_resume']
MEMBERS = ['world.bin', 'manifest.json', 'thumbnail.png', 'journal.log']
WRITER = ['unsupported', 'out_of_bounds', 'supported_connected', 'overlap',
'missing_connection', 'lower_stair', 'upper_stair', 'refused_demolition',
'cancel_demolition', 'demolition', 'rebuild', 'pan_zoom']
RUNTIME_KEYS = set(('runtime_schema_version weather world_seed next_command_id '
'staffing_projection_enabled save_lifecycle command_queue id_generator domain_events rng_streams '
'world path movement_credits mana economy facility_upkeep upkeep_settlement inventory kitchen_production '
'auto_purchase_plans elevators legacy_stairs logistics_dispatcher logistics_requests facilities '
'facility_cleanliness guests staff catalog_manifest progression hired_staff_candidate_ids operations '
'relationships event_definitions events authority defense construction excavation spatial_subslots '
'route_projection facility_infrastructure').split())
REASONS = {'unsupported': ('UNSUPPORTED', '配置不可: 構造を支えられません'),
'out_of_bounds': ('OUT_OF_BOUNDS', '配置不可: 世界の範囲外です'),
'overlap': ('OVERLAP', '配置不可: ほかの施設または工事と重なります'),
'resume_overlap': ('OVERLAP', '配置不可: ほかの施設または工事と重なります')}
def module(name, path):
spec = importlib.util.spec_from_file_location(name, path)
obj = importlib.util.module_from_spec(spec)
spec.loader.exec_module(obj)
return obj
prior = module('p08_placement_typed_codec', HERE/'check_p08_migration_evidence.py')
runner = module('p08_placement_source_inventory', HERE/'run_all_tests.py')
decode = prior.decode
typed_differences = prior.typed_differences
def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def load(path): return json.loads(Path(path).read_bytes())
def require(ok, message):
if not ok: raise ValueError(message)
def png(data):
require(data.startswith(b'\x89PNG\r\n\x1a\n'), 'missing OS screenshot')
offset = 8; chunks = []; payload = b''
while offset < len(data):
require(offset+12 <= len(data), 'invalid OS screenshot')
size = int.from_bytes(data[offset:offset+4], 'big')
kind = data[offset+4:offset+8]; body = data[offset+8:offset+8+size]
require(offset+12+size <= len(data), 'invalid OS screenshot')
crc = int.from_bytes(data[offset+8+size:offset+12+size], 'big')
require(zlib.crc32(kind+body) == crc, 'invalid OS screenshot CRC')
if kind == b'IHDR':
require(len(body) == 13 and all(v > 0 for v in struct.unpack('>II',body[:8])), 'invalid OS screenshot dimensions')
if kind == b'IDAT': payload += body
chunks.append(kind); offset += size+12
require(chunks and chunks[0] == b'IHDR' and chunks[-1] == b'IEND' and payload, 'invalid OS screenshot chunks')
try:
require(bool(zlib.decompress(payload)), 'empty OS screenshot')
except zlib.error as exc:
raise ValueError('invalid OS screenshot pixels') from exc
def input_map(root):
names = [SPEC, CONTRACT, 'AGENTS.md', RECORDER, RECORDER+'.uid',
'docs/plans/original-completion-2026-10-02/IMPLEMENTATION_PLAN.md',
'docs/traceability/original/BASELINE.txt']
names += [p.relative_to(root).as_posix() for p in (root/'tests/fixtures/p08-03').rglob('*')
if p.is_file() and '__pycache__' not in p.parts and '.scratch' not in p.parts]
return {name: sha(root/name) for name in sorted(names)}
def keyed(rows, key):
result = {r[key]: r for r in rows}
require(len(result) == len(rows), 'duplicate '+key)
return result
def state(node):
snap = decode(node)
require(set(snap) == {'schema_version', 'clock', 'game_state', 'systems'}, 'missing whole snapshot envelope')
r = snap['systems']['runtime']
require(set(r) == RUNTIME_KEYS, 'missing whole Runtime members')
facilities = keyed(r['construction']['facilities'], 'instance_id')
jobs = keyed(r['construction']['jobs'], 'job_id')
for job in jobs.values():
if job['state'] == 'COMPLETED':
require(job['facility_instance_id'] in facilities, 'orphan completed job')
owners = {k: v['occupied_cells'] for k, v in facilities.items()}
owners.update({k: v['reserved_cells'] for k, v in jobs.items() if v['state'] not in ('COMPLETED', 'CANCELLED')})
occupancy = keyed(r['world']['occupancies'], 'entity_id')
require(set(owners) == set(occupancy), 'orphan/lost occupancy')
seen = set()
for key, cells in owners.items():
actual = {tuple(c) for c in occupancy[key]['cells']}
require(actual == {tuple(c) for c in cells} and len(actual) == len(cells), 'partial occupancy')
require(not seen.intersection(actual), 'overlapping occupancy')
seen.update(actual)
operational = keyed(r['facilities']['instances'], 'instance_id')
require(set(operational) == set(facilities), 'partial demolition/completion')
for key, row in operational.items():
require(row['occupied_cells'] == facilities[key]['occupied_cells'], 'operational cells drift')
return snap
def coordinates(p):
cell = p['world_cell']
require(cell == p['selected_cell'] and len(cell) == 3, 'world-cell drift')
bounds, origin, dimensions = p['bounds'], p['map_position'], p['cell_dimensions']
require(len(dimensions) == 2 and all(math.isfinite(v) and v > 0 for v in dimensions), 'invalid cell dimensions')
wanted = [origin[0] + (cell[0]-bounds['min_x']+0.5)*dimensions[0],
origin[1] + (bounds['max_y']-cell[1]+0.5)*dimensions[1]]
require(all(math.isfinite(v) for v in p['viewport']+p['local']), 'nonfinite viewport')
require(all(abs(a-b) < 0.001 for a, b in zip(p['local'], wanted)), 'converted cell pixel mismatch')
require(0.5 <= p['zoom'] <= 4.0, 'zoom bounds')
for target in ('viewport','screen'):
if target == 'screen' and target not in p: continue
matrix = p[target+'_transform']
require(len(matrix) == 6 and all(math.isfinite(v) for v in matrix), 'invalid viewport transform')
x, y = p['local']
expected = [matrix[0]*x+matrix[2]*y+matrix[4], matrix[1]*x+matrix[3]*y+matrix[5]]
require(len(p[target]) == 2 and all(abs(a-b) < 0.001 for a,b in zip(expected,p[target])), 'transformed '+target+' mismatch')
def preserve(a, b, results):
differences = typed_differences(a, b)
if not results:
require(not differences, 'cancel/preview partially applied')
return
# Keep every typed field. Only the command envelope may grow on a refusal.
allowed = [('systems', 'runtime', key) for key in
('command_queue', 'next_command_id', 'domain_events', 'id_generator')]
require(all(any(p[:3] == prefix for prefix in allowed) for p in differences), 'refusal partially applied')
old, new = [decode(n)['systems']['runtime'] for n in (a, b)]
require(new['next_command_id'] == old['next_command_id'] + len(results), 'refusal command sequence')
oq, nq = old['command_queue'], new['command_queue']
require(nq['processed'] == oq['processed'] + results, 'refusal result/journal mismatch')
for key in ('pending', 'in_flight'):
require(nq[key] == oq[key], 'refusal queue loss')
def action_gate(row, typed):
coordinates(row['position'])
before, after = typed(row['before']), typed(row['after'])
a, b = state(before), state(after)
require(row['position']['bounds'] == a['systems']['runtime']['world']['bounds'], 'invented world bounds')
kind, label = row['kind'], row['id']
expected_kind = 'camera' if label == 'pan_zoom' else 'cancel' if label == 'cancel_demolition' else 'demolish' if label in ('demolition','refused_demolition') else 'build'
require(kind == expected_kind, 'operation kind mismatch')
expected_reason = REASONS[label][0] if label in REASONS else {'missing_connection':'STAIR_LANDING_MISSING', 'refused_demolition':'STAIR_LANDING_IN_USE'}.get(label, '')
require(row.get('expected', expected_reason) == expected_reason, 'required refusal reason changed')
results = row.get('results', [])
if label in REASONS:
code, message = REASONS[label]
require(row['report']['error_codes'][0] == code and row['status'] == message, 'reason mismatch')
require(not row['report']['can_build'] and not results, 'preview rejection bypassed')
preserve(before, after, results)
elif kind == 'cancel':
require(not results and row['status'] == '撤去を取り消しました。', 'cancel submitted command')
preserve(before, after, results)
elif kind == 'camera':
preserve(before, after, [])
require(row['pan_before'] != row['pan_after'], 'move-mode wheel did not pan')
require(row['ctrl_pressed'] is False and row['steps'] == [max(0.5, row['steps'][0]-0.5*i) for i in range(9)] + [min(4.0, 1.0+0.5*i) for i in range(9)], 'wheel clamp sequence')
require(row['steps'][8] == 0.5 and row['steps'][-1] == 4.0, 'missing clamp boundaries')
require(row['original_local'] != row['position']['local'], 'fixed pixels substituted for world identity')
else:
require(len(results) == 1, 'one real Runtime result required')
result = results[0]
processed = b['systems']['runtime']['command_queue']['processed']
require(result in processed, 'result absent from actual model')
if row['expected']:
require(result['code'] == 'VALIDATION_FAILED' and result['details']['reason'] == row['expected'], 'reason mismatch')
preserve(before, after, results)
else:
require(result['code'] == 'OK', 'positive rejected')
if kind == 'demolish':
require(row['status'] == '%s | %s | %s' % (result['code'], result['details'].get('reason', ''), result['message']), 'displayed domain reason mismatch')
if not row['expected']:
fid = row['instance_id']
old = keyed(a['systems']['runtime']['construction']['facilities'], 'instance_id')
new = keyed(b['systems']['runtime']['construction']['facilities'], 'instance_id')
require(fid in old and fid not in new and set(new) == set(old)-{fid}, 'partial demolition')
require(row['position']['world_cell'] in old[fid]['occupied_cells'], 'demolition wrong cell')
refund = result['details']['refund']
resources_a, resources_b = a['game_state']['resources'], b['game_state']['resources']
require(set(resources_a) == set(resources_b) and all(resources_b[k] == v + refund.get(k, 0) for k, v in resources_a.items()), 'demolition refund mismatch')
if kind == 'build' and not row['expected']:
completed = state(typed(row['completed']))
facilities = completed['systems']['runtime']['construction']['facilities']
require(any(f['definition_id'] == row['definition'] and f['origin'] == row['position']['world_cell'] for f in facilities), 'construction completion missing')
require(row['report']['can_build'], 'positive support invalid')
cost = row['report']['total_cost']
resources_a, resources_b = a['game_state']['resources'], b['game_state']['resources']
require(set(resources_a) == set(resources_b) and all(resources_b[k] == v - cost.get(k, 0) for k, v in resources_a.items()), 'build cost mismatch')
if label == 'supported_connected':
require(row['report']['connection']['status'] == 'CONNECTED', 'positive connection absent')
if label == 'missing_connection':
require(row['status'] == '操作に失敗しました: '+result['message'], 'connection reason mismatch')
require(row['report']['can_build'], 'connection refusal did not reach Runtime')
def check(project_root, evidence_dir):
root, evidence = Path(project_root).resolve(), Path(evidence_dir).resolve()
try:
index = load(evidence/'index.json')
require(index['schema'] == 'p08-normal-placement-v1' and index['acceptance'] == 'pending', 'schema/acceptance')
require(index['collection_error'] is None, 'collection failed')
require(index['conditions'] == CONDITIONS, 'missing condition')
actual = {n: v['sha256'] for n, v in runner.source_refs(root).items()}
require(index['source_map'] == actual, 'stale source')
require(index['inputs'] == input_map(root), 'stale input')
require(index['spec_sha256'] == SPEC_SHA == hashlib.sha256((root/SPEC).read_text().strip().encode()).hexdigest(), 'immutable spec')
require(index['marker_contract_sha256'] == sha(root/CONTRACT), 'stale marker')
require(re.fullmatch('[0-9a-f]{40}', index['commit']) is not None, 'commit')
require(index['commit'] == subprocess.check_output(['git','-C',str(root),'rev-parse','HEAD'],text=True).strip(), 'stale commit')
def artifact(name):
require(isinstance(name, str) and name and not Path(name).is_absolute() and '..' not in Path(name).parts, 'unsafe/missing artifact')
p = evidence/name
require(not p.is_symlink() and p.resolve().is_relative_to(evidence), 'artifact escape')
pin = index['artifacts'][name]
require(p.stat().st_size == pin['bytes'] and sha(p) == pin['sha256'], 'artifact hash '+name)
return p
def document(name): return load(artifact(name))
for name in index['artifacts']: artifact(name)
run = document('runner-input.json')
require(run['selected_source_map'] == actual and run['commit'] == index['commit'], 'runner source')
require(sha(run['engine']) == run['engine_sha256'], 'engine hash')
ended = document('runner-ended.json')
require(ended['error'] is None and not ended['cleanup_errors'], 'outer failure')
markers = module('p08_placement_markers', HERE/'strict_test_markers.py')
contracts = markers.load_contracts(root/CONTRACT)
pids = set()
def phase(name):
receipt = document(name)
pid = receipt['actual_pid']
require(type(pid) is int and pid > 0 and pid not in pids, 'missing/duplicate actual PID')
pids.add(pid)
require('--headless' in receipt['command'] and '--script' in receipt['command']
and '--path' in receipt['command'], 'handler command contract')
prefix = str(Path(name).parent)+'/'
folder = prefix+receipt['phase']+'/'
start, end, outer = [document(folder+n) for n in ('started.json', 'ended.json', 'outer-owned-finally.json')]
require(all(v['actual_pid'] == pid and v['command'] == receipt['command'] for v in (start, end)), 'raw lifecycle identity')
require(start.get('live_identity_exact') is True or start.get('tracked_child_ended_before_initial_probe') is True, 'unobserved native PID')
require(end['engine_ended'] is True and end['exit_code'] == 0, 'unended PID')
require(outer['actual_captured_pids'] == [pid] and outer['all_captured_ended'] is True and not outer['outer_cleanup_errors'] and outer['foreign_process_or_group_signals'] == 0, 'owned cleanup')
native = receipt['native_process']
require(native['exit_code'] == native['helper_exit_code'] == 0 and native['timed_out'] is False and native['lifecycle_error'] is None, 'native failure')
require(receipt['exit_code'] == 0 and receipt['engine_ended'] is True and not receipt['failed'], 'phase failure')
profile = receipt['actual_private_profile']['report']
require(profile['fresh_before_parent_initialize'] is True and profile['profile_writes_before_guard'] == 0 and profile['actual_user_data_dir'] == profile['expected_user_data_dir'], 'profile guard')
require(receipt['test'] == TEST and receipt['source_sha256'] == actual[TEST], 'executed suite source')
require(not markers.parse_raw(TEST, artifact(prefix+receipt['log']).read_text(), contracts, actual[TEST])['errors'], 'strict completion')
return pid
def os_phase(name):
end = document(name)
prefix = str(Path(name).parent)+'/'
start = document(prefix+'started.json')
spawned = document(prefix+'process-started.json')
pid = end['actual_pid']
require(type(pid) is int and pid > 0 and pid not in pids, 'missing/duplicate actual PID')
pids.add(pid)
command = end['command']
require(command == [run['engine'], '--path', end['stage'], '--script', command[-1]], 'OS native argv')
require('--headless' not in command and len(command) == 5, 'OS headless masquerade')
require(spawned['actual_pid'] == pid and spawned['command'] == command and spawned['monotonic_ns'] > 0,
'OS raw process identity')
for doc in (start, spawned, end):
require(doc['protocol'] == 'p08-placement-native-os-v1' and doc['source_map'] == actual
and doc['commit'] == index['commit'], 'OS exact source/commit')
require(doc['controller_sha256'] == actual['tools/run_p08_normal_placement.py']
and doc['recorder_sha256'] == sha(root/RECORDER), 'OS controller/recorder SHA')
require(doc['command'] == command and doc['engine_sha256'] == run['engine_sha256'], 'OS command/engine drift')
require(sha(artifact(prefix+'passive_os.gd')) == sha(root/RECORDER), 'executed OS recorder SHA')
require(end['error'] is None and not end['guard_errors'] and not end['cleanup_errors']
and end['exit_code'] == 0 and end['engine_ended'] is True and end['passed'] is True, 'OS native failure')
cleanup = end['owned_cleanup']
require(len(cleanup) == 1 and cleanup[0]['pid'] == pid and cleanup[0]['engine_ended'] is True
and not cleanup[0]['outer_cleanup_errors'] and cleanup[0]['foreign_process_or_group_signals'] == 0,
'OS owned cleanup')
require(end['stage_before'] == end['stage_after'] and end['default_before'] == end['default_after'], 'OS protected state changed')
require({n:v['sha256'] for n,v in end['source_after'].items()} == actual, 'OS source changed')
require(end['profile_before'] == {'existed':False,'files':{}}, 'OS profile not fresh')
owners = module('p08_verify_os_owners', HERE/'regression/ordinary_proof_gate.py')
actual_owner_proof = owners.ordinary_gate(evidence/prefix/'teardown', pid, evidence/prefix, ['single'])
require(end['ordinary_owner_proof'] == actual_owner_proof, 'OS ordinary owner proof mismatch')
profile = document('os-pair/'+end['role']+'-profile.json')
require(profile['pid'] == pid and Path(profile['user_data_dir']).resolve() == Path(end['profile']).resolve()
and profile['display'] != 'headless', 'OS actual profile/display')
raw = artifact(prefix+'engine-native.log').read_text()
require('ERROR:' not in raw and re.findall(r'(?m)^P08 passive OS placement: ([0-9]+) checks, 0 failures$', raw)
and 'P08 normal placement:' not in raw, 'OS raw diagnostics/completion')
members = end['formal_input']
require(set(members) == set(MEMBERS), 'OS missing formal input')
source_dir = 'pair/initial-formal' if end['role'] == 'writer' else 'os-pair/formal'
for member, pin in members.items():
source = artifact(source_dir+'/'+member)
require(pin == {'bytes':source.stat().st_size,'sha256':sha(source)}, 'OS formal input SHA')
return pid
case = index['cases'][0]
require(len(index['cases']) == 1 and len(case['phases']) == 2, 'writer/reader pair')
pair_pids = [phase(n) for n in case['phases']]
def pair_gate(directory, expected_pids, native_os=False):
def typed(name):
artifact(directory+'/'+name+'.var')
return document(directory+'/'+name+'.json')
rows = []
for role, pid in zip(('writer', 'reader'), expected_pids):
process = document(directory+'/'+role+'-process.json')
require(process['pid'] == pid and process['failures'] == 0 and process['checks'] > 0, 'process attribution')
require(bool(process['engine']['hash']) and bool(process['engine']['string']), 'engine version')
doc = document(directory+'/'+role+'-actions.json')
require(doc['setup'] == 'diagnostic', 'unlabelled diagnostic')
expected = WRITER if role == 'writer' else ['resume_overlap']
require([r['id'] for r in doc['actions']] == expected, 'missing operations')
if role == 'writer':
budget = doc['budget']
require(budget['setup'] == 'diagnostic', 'budget setup')
costs = {'reception': root/'data/facilities/reception.tres', 'guest_stairs': root/'data/facilities/infrastructure/guest_stairs.tres'}
total = {'gold': 0, 'materials': 0}
require([r['definition'] for r in budget['costs']] == ['reception', 'reception', 'guest_stairs', 'guest_stairs'], 'diagnostic build roster')
for entry in budget['costs']:
text = costs[entry['definition']].read_text()
line = re.search(r'^construction_cost = (.+)$', text, re.M).group(1)
original = {k: int(v) for k, v in re.findall(r'&"([^"]+)":\s*(\d+)', line)}
require(entry['cost'] == original, 'invented diagnostic cost')
for k in total: total[k] += original.get(k, 0)
delta = {k: v-budget['initial'][k] for k, v in total.items() if v > budget['initial'][k]}
require(total == budget['needed'] and delta == budget['deltas'], 'diagnostic budget mismatch')
for row in doc['actions']:
require(row['input_method'] == ('native_os' if native_os else 'handler'), 'fake OS provenance')
action_gate(row, typed)
if row['kind'] == 'build':
wanted_definition = 'standard_room' if row['id'] == 'unsupported' else 'guest_stairs' if row['id'] in ('missing_connection','lower_stair','upper_stair') else 'reception'
require(row['definition'] == wanted_definition, 'diagnostic definition substituted')
if not row.get('expected', REASONS.get(row['id'], ('',))[0]):
path = root/('data/facilities/infrastructure/' if wanted_definition == 'guest_stairs' else 'data/facilities/')/(wanted_definition+'.tres')
text = path.read_text()
line = re.search(r'^construction_cost = (.+)$',text,re.M).group(1)
cost = {k:int(v) for k,v in re.findall(r'&"([^"]+)":\s*(\d+)',line)}
require(row['report']['total_cost'] == cost, 'invented production cost')
width,height = map(int,re.search(r'^footprint = Vector2i\((\d+), (\d+)\)$',text,re.M).groups())
x,y,z = row['position']['world_cell']
cells = {(x+dx,y+dy,z) for dx in range(width) for dy in range(height)}
completed = state(typed(row['completed']))['systems']['runtime']['construction']['facilities']
matches = [f for f in completed if f['origin'] == [x,y,z] and f['definition_id'] == wanted_definition]
require(len(matches) == 1 and {tuple(c) for c in matches[0]['occupied_cells']} == cells, 'production footprint mismatch')
rows += doc['actions']
checkpoint = typed('writer-checkpoint')
require(checkpoint == typed('reader-checkpoint'), 'wrong resume model')
state(checkpoint)
require(sha(artifact(directory+'/writer-checkpoint.var')) == sha(artifact(directory+'/reader-checkpoint.var')), 'wrong raw resume')
saved = document(directory+'/save.json')
require(set(saved['members']) == set(MEMBERS) and saved['pid'] == expected_pids[0], 'missing formal members/writer')
for member in MEMBERS:
require(saved['members'][member] == sha(artifact(directory+'/formal/'+member)), 'formal member changed')
facilities = keyed(decode(checkpoint)['systems']['runtime']['construction']['facilities'], 'instance_id')
require(saved['instance_id'] in facilities and facilities[saved['instance_id']]['origin'] == saved['cell'], 'wrong resume instance')
by_id = {r['id']: r for r in rows}
require(all(by_id[n]['position']['world_cell'] == saved['cell'] for n in ('supported_connected', 'demolition', 'rebuild', 'pan_zoom', 'resume_overlap')), 'target world-cell drift')
require(by_id['demolition']['instance_id'] != saved['instance_id'], 'rebuild did not create new instance')
return rows
pair_gate(case['directory'], pair_pids)
os_proof = document(index['os_proof'])
require(os_proof['source_map'] == actual and os_proof['inputs'] == index['inputs'], 'stale OS source/input')
require(os_proof['input_method'] == 'native_os' and os_proof['observer_mutations'] == [], 'fake OS provenance')
os_pids = [os_phase(n) for n in os_proof['phases']]
require(len(os_pids) == 2, 'OS separate PID pair')
os_rows = pair_gate(os_proof['directory'], os_pids, True)
lifecycle = []
for role, pid in zip(('writer','reader'),os_pids):
requests = document('os-pair/'+role+'-requests.json')
require(requests['pid'] == pid, 'OS lifecycle PID')
actions = [document(str(Path(name).parent)+'/request.json') for name in requests['requests']]
require(any(a['action'] == 'load_menu' for a in actions), 'missing OS normal load')
if role == 'writer':
require(any(a['action'] == 'button' and a.get('node','').endswith('/ManualSaveButton') for a in actions), 'missing OS normal save')
lifecycle.append({'id':role+'-lifecycle', 'requests':requests['requests']})
all_ack = {name for name in index['artifacts'] if name.startswith('os-pair/') and name.endswith('/ack.json')}
require(all_ack == {name for row in lifecycle for name in row['requests']}, 'omitted OS lifecycle requests')
require(all(set(row['requests']) <= all_ack for row in os_rows), 'OS action request missing')
os_rows += lifecycle
for row in os_rows:
requests = row['requests']
require(requests and len(requests) == len(set(requests)), 'missing/duplicate OS requests')
expected_pid = os_pids[1 if row['id'] in ('resume_overlap','reader-lifecycle') else 0]
wheel_requests = []
for name in requests:
ack = document(name)
folder = str(Path(name).parent)+'/'
request = document(folder+'request.json')
observed = document(folder+'observed.json')
require(ack['target_pid'] == request['target_pid'] == observed['pid'] == expected_pid
and ack['sequence'] == request['sequence'] == observed['sequence'], 'fake OS provenance')
require(ack['request_sha256'] == sha(artifact(folder+'request.json')), 'OS ack request mismatch')
raw = document(ack['input_receipt'])
require(raw['method'] == 'native_os' and raw['target_pid'] == expected_pid and raw['timestamp_ns'] > 0
and raw['tool_name'] and raw['tool_call_id'], 'fake OS provenance')
require(raw['events'] and observed['events'], 'missing OS events')
require(raw['collector_pid'] > 0 and raw['collector_pid'] not in os_pids, 'fake OS collector')
for key in ('screen_before', 'screen_after'):
png(artifact(ack[key]).read_bytes())
if request['action'] == 'cell' and 'position' in row:
require(request['world_cell'] == row['position']['world_cell']
and request['viewport'] == row['position']['viewport']
and request['screen'] == row['position']['screen'], 'OS transformed cell drift')
if request['action'] == 'wheel':
wheel_requests.append(request)
if row['id'] == 'pan_zoom':
require(any(e['move_mode'] and not e['ctrl'] for e in wheel_requests), 'missing OS pan')
require(sum(not e['move_mode'] and not e['ctrl'] for e in wheel_requests) == 18, 'missing OS zoom')
# Full regression remains a separately verified Root requirement; this
# finite gate cannot replace the standard runner or independent review.
return []
except (KeyError, ValueError, TypeError, OSError, IndexError, AttributeError, AssertionError, subprocess.SubprocessError) as exc:
return [str(exc)]
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--verify-only', action='store_true', required=True)
parser.add_argument('--project-root', type=Path, default=HERE.parent)
args = parser.parse_args(argv)
build = os.environ.get('BUILD_OUTPUT_DIR')
if not build:
print('BUILD_OUTPUT_DIR is required', file=sys.stderr)
return 2
deficits = check(args.project_root, Path(build)/'p08-normal-placement-evidence')
print(json.dumps({'task_id': 'P08-03', 'integrity': 'fail' if deficits else 'pass',
'deficits': deficits, 'acceptance': 'pending'}, ensure_ascii=False))
return 1 if deficits else 0
if __name__ == '__main__':
raise SystemExit(main())
tools/regression/suite-policy.json
+9
View File
@@ -391,6 +391,15 @@
],
"old_bounded_teardown": false,
"new_audio_QA": false
},
"tests/ui/test_p08_normal_placement_ops.gd": {
"source_sha256": "734f4f85d0b0fc45dc31fc23502425b911635667dabc0341ba0d88b0107f4405",
"ordinary_cleanup_prepared": true,
"ordinary_control_labels": [
"single"
],
"old_bounded_teardown": false,
"new_audio_QA": false
}
},
"outside_routes": {
tools/run_p08_normal_placement.py
+248
View File
@@ -0,0 +1,248 @@
#!/usr/bin/env python3
"""Root-only native collection. Does not run regression or claim acceptance."""
import argparse
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import time
import uuid
import re
sys.dont_write_bytecode = True
HERE = Path(__file__).resolve().parent
CONTRACT = 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json'
CONDITIONS = ('support_connection', 'invalid_position_reason', 'demolish_rebuild',
'pan_zoom_world_identity', 'separate_pid_resume')
RECORDER = 'docs/plans/original-completion-2026-10-02/execution/p08-03/passive_os_placement_recorder.gd'
def collect_os(r, root, out, stage, engine, commit, refs, protected, outer):
"""P08-02 collect_native_os protocol; deliberately not execute_phase/headless.
Root supplies CUA ack/receipt files in the requested directories. Neither
controller nor passive endpoints generate input or change permissions.
"""
r.require(not outer.native.helper['live_engines'](), 'external engine present')
before = r.inventory(stage, stage=True)
public = (stage/'project.godot').read_bytes()
pair = out/'os-pair'
pair.mkdir()
r.copy_source(out/'pair/diagnostic-budget.json', pair/'diagnostic-budget.json')
phases = []
for role in ('writer', 'reader'):
r.require(not outer.native.helper['live_engines'](), 'external engine present')
r.require(r.source_refs(root) == refs and r.default_player(root) == protected, 'OS before guards')
r.require(r.inventory(stage, stage=True) == before, 'OS stage before guard')
r.require(shutil.disk_usage(out).free >= 500*1024*1024, 'OS disk admission')
folder = out/'os-runs'/role
folder.mkdir(parents=True)
name = 'TheTree-PCK-tests-'+uuid.uuid4().hex+'-p08-placement-'+role
profile = r.profile_path(name)
r.require(not profile.exists(), 'fresh OS private profile')
formal_source = out/('pair/initial-formal' if role == 'writer' else 'os-pair/formal')
formal_before = r.inventory(formal_source)
r.require(set(formal_before) == {'world.bin','manifest.json','thumbnail.png','journal.log'}, 'all formal inputs')
formal = profile/'saves/manual_ui'
formal.mkdir(parents=True)
for member, pin in formal_before.items():
r.copy_source(formal_source/member, formal/member)
r.require(r.sha(formal/member) == pin['sha256'], 'formal fixture copy SHA')
recorder = folder/'passive_os.gd'
r.copy_source(root/RECORDER, recorder)
proof = folder/'teardown'
(proof/'owned-audio').mkdir(parents=True)
env = r.fresh_environment(dict(os.environ))
env.update(P08_PLACEMENT_ROOT=str(pair), P08_PLACEMENT_ROLE=role,
THE_TREE_AUDIO_TEARDOWN_PROOF_DIR=str(proof/'owned-audio'),
THE_TREE_AUDIO_FIXTURE_CONTROL_PROOF=str(proof/'control.json'))
command = [engine, '--path', str(stage), '--script', str(recorder)]
base = {'protocol':'p08-placement-native-os-v1', 'role':role, 'commit':commit,
'source_map':{n:v['sha256'] for n,v in refs.items()}, 'stage':str(stage),
'stage_before':before, 'profile':str(profile), 'profile_name':name,
'profile_before':{'existed':False, 'files':{}}, 'formal_input':formal_before,
'formal_source':str(formal_source), 'owner_pid':os.getpid(),
'default_before':protected, 'command':command, 'engine_sha256':r.sha(engine),
'controller_sha256':r.sha(root/'tools/run_p08_normal_placement.py'),
'recorder_sha256':r.sha(recorder), 'formal_acceptance':False,
'environment':{k:v for k,v in env.items() if k.startswith(('P08_', 'THE_TREE_'))},
'environment_sha256':hashlib.sha256(json.dumps(env,sort_keys=True).encode()).hexdigest()}
r.write(folder/'started.json', base)
error = None; children = []; guards = []; exit_code = None; owner_proof = None
try:
(stage/'project.godot').write_bytes(r.private_project(public, name))
with (folder/'engine-native.log').open('x') as log:
process = subprocess.Popen(command, cwd=stage, env=env, stdout=log, stderr=subprocess.STDOUT)
children.append(process)
r.write(folder/'process-started.json', dict(base, actual_pid=process.pid,
monotonic_ns=time.monotonic_ns(), state='waiting_for_root_cua_requests'))
print(json.dumps({'actual_pid':process.pid, 'request_directory':str(pair),
'state':'waiting_for_root_cua', 'role':role}), flush=True)
exit_code = process.wait(timeout=900)
r.require(exit_code == 0, 'ordinary native OS exit0')
raw = (folder/'engine-native.log').read_text()
r.require('ERROR:' not in raw and 'SCRIPT ERROR:' not in raw, 'OS native diagnostics')
matches = re.findall(r'(?m)^P08 passive OS placement: ([0-9]+) checks, 0 failures$', raw)
r.require(len(matches) == 1 and int(matches[0]) > 0, 'passive observer completion')
observed = r.read(pair/(role+'-profile.json'))
r.require(observed['pid'] == process.pid and Path(observed['user_data_dir']).resolve() == profile.resolve()
and observed['display'] != 'headless', 'actual OS private profile/display')
owners = r.module('p08_os_owner_gate', r.SUPPORT/'ordinary_proof_gate.py')
owner_proof = owners.ordinary_gate(proof, process.pid, folder, ['single'])
except BaseException as caught:
error = type(caught).__name__+': '+str(caught)
finally:
try:
cleanup, cleanup_errors = outer.finish_captured(children)
finally:
(stage/'project.godot').write_bytes(public)
try:
r.require(r.source_refs(root) == refs, 'OS source unchanged')
r.require(r.inventory(stage, stage=True) == before, 'OS stage unchanged')
r.require(r.default_player(root) == protected, 'OS default player unchanged')
r.require(r.inventory(formal_source) == formal_before, 'formal input unchanged')
r.require(r.sha(recorder) == base['recorder_sha256'], 'recorder unchanged')
except BaseException as caught:
guards.append(type(caught).__name__+': '+str(caught))
passed = not (error or guards or cleanup_errors) and len(children) == 1 and exit_code == 0
r.write(folder/'ended.json', dict(base, actual_pid=children[0].pid if children else None,
exit_code=exit_code, engine_ended=bool(cleanup) and all(v['engine_ended'] for v in cleanup),
error=error, guard_errors=guards, cleanup_errors=cleanup_errors, owned_cleanup=cleanup,
ordinary_owner_proof=owner_proof, stage_after=r.inventory(stage,stage=True),
default_after=r.default_player(root), source_after=r.source_refs(root),
profile_after=r.inventory(profile), passed=passed))
phases.append('os-runs/'+role+'/ended.json')
r.require(passed, 'native OS collection failed: '+str(error or guards or cleanup_errors))
return phases
def module(name, path):
spec = importlib.util.spec_from_file_location(name, path)
obj = importlib.util.module_from_spec(spec)
spec.loader.exec_module(obj)
return obj
def absolute(value):
path = Path(value)
if not path.is_absolute():
raise argparse.ArgumentTypeError('absolute path required')
return path.resolve()
def collect(args):
root, out = args.project_root, args.output
if out.is_relative_to(root):
raise ValueError('evidence output must be outside the frozen source checkout')
r = module('p08_standard_runner', root/'tools/run_all_tests.py')
commit = subprocess.check_output(['git', '-C', str(root), 'rev-parse', 'HEAD'], text=True).strip()
r.require(commit == args.commit, 'actual final submitted commit required')
r.require(not subprocess.check_output(['git','-C',str(root),'status','--porcelain'], text=True), 'frozen clean worktree required')
r.require(not out.exists(), 'fresh output; never overwrite earlier proof')
refs = r.source_refs(root)
checker = module('p08_initial_inputs', root/'tools/check_p08_normal_placement_evidence.py')
inputs_before = checker.input_map(root)
outer = r.module('p08_owned_outer', r.SUPPORT/'owned_outer.py')
r.require(not outer.native.helper['live_engines'](), 'one native engine')
engine = shutil.which(args.godot)
r.require(engine is not None, 'native engine exists')
engine = str(Path(engine).resolve())
out.mkdir(parents=True)
stage = out/'stage'; stage.mkdir()
(out/'starters').mkdir(); (out/'bootstrap').mkdir()
for name, pin in r.read(r.SUPPORT/'bootstrap-scripts.json').items():
data = pin['text'].encode()
r.require(hashlib.sha256(data).hexdigest() == pin['sha256'], 'bootstrap pin')
(out/'bootstrap'/name).write_bytes(data)
for name, pin in refs.items():
target = stage/name; target.parent.mkdir(parents=True, exist_ok=True)
r.copy_source(Path(pin['path']), target)
r.require(r.sha(target) == pin['sha256'], 'whole copied source '+name)
expected = {n: {'bytes':v['bytes'], 'sha256':v['sha256']} for n,v in refs.items()}
selected = {'root_before':refs, 'map':{n:v['sha256'] for n,v in refs.items()}, 'timeout':180}
r.require(r.inventory(stage, stage=True) == expected, 'complete private source')
protected = r.default_player(root); public = (stage/'project.godot').read_bytes()
paths = ['res://'+n for n in sorted(refs) if n.endswith('.gd')]
r.write(out/'parse-paths.json', paths)
markers = r.module('p08_strict_markers', root/'tools/strict_test_markers.py')
contracts = markers.load_contracts(root/CONTRACT)
policy = r.read(r.SUPPORT/'suite-policy.json')
diagnostics = r.module('p08_diagnostics', r.SUPPORT/'diagnostic_gate.py')
owners = r.module('p08_owners', r.SUPPORT/'ordinary_proof_gate.py')
audio = r.module('p08_audio', r.SUPPORT/'audio_evidence_gate.py')
r.write(out/'runner-input.json', {'protocol':'p08-normal-placement-v1', 'commit':commit,
'selected_source_map':selected['map'], 'engine':engine, 'engine_sha256':r.sha(engine),
'owner_pid':os.getpid(), 'default_player':protected, 'diagnostic_seed':42533, 'inputs':inputs_before})
phases = []; previous = {}; cases = []; error = None; os_proof = None
deadline = time.monotonic()+(7200 if args.input_mode == 'native_os' else 900)
saved_env = {k:os.environ.get(k) for k in ('P08_PLACEMENT_ROOT','P08_PLACEMENT_ROLE','P08_PLACEMENT_INPUT')}
def phase(label, test):
remaining = int(deadline-time.monotonic())
r.require(remaining >= (180 if label == 'import' else 1), '900s collection deadline')
selected['timeout'] = min(180, remaining)
phase_out = out/'runs'/label
phase_out.mkdir(parents=True)
(phase_out/'starters').mkdir(); (phase_out/'bootstrap').mkdir()
for original in (out/'bootstrap').iterdir():
r.copy_source(original,phase_out/'bootstrap'/original.name)
r.require(r.sha(original)==r.sha(phase_out/'bootstrap'/original.name),'bootstrap full copy')
r.write(phase_out/'parse-paths.json',paths)
row = r.execute_phase(engine, stage, phase_out, label, test, selected, expected, previous,
policy, contracts, paths, public, protected, root, outer, markers, diagnostics, owners, audio)
phases.append('runs/'+label+'/'+label+'-receipt.json')
r.require(not row['failed'], 'strict phase failed: '+label)
try:
phase('import', None); phase('parse', None)
pair = out/'pair'
preserved = {}
for role in ('writer', 'reader'):
os.environ.update(P08_PLACEMENT_ROOT=str(pair), P08_PLACEMENT_ROLE=role, P08_PLACEMENT_INPUT='handler')
phase(role, 'tests/ui/test_p08_normal_placement_ops.gd')
actual = r.inventory(pair)
r.require(all(actual.get(n) == pin for n, pin in preserved.items()),
'reader preserved every writer and formal byte')
preserved = actual
cases.append({'directory':'pair', 'phases':phases[-2:], 'files':preserved})
if args.input_mode == 'native_os':
os_phases = collect_os(r, root, out, stage, engine, commit, refs, protected, outer)
checker = module('p08_os_input_map', root/'tools/check_p08_normal_placement_evidence.py')
os_proof = 'os-proof.json'
r.write(out/os_proof, {'source_map':selected['map'], 'inputs':checker.input_map(root),
'input_method':'native_os', 'observer_mutations':[], 'phases':os_phases, 'directory':'os-pair'})
except BaseException as caught:
error = type(caught).__name__+': '+str(caught)
finally:
cleanup, cleanup_errors = outer.finish_captured(outer.OWNED)
for key,value in saved_env.items():
if value is None: os.environ.pop(key, None)
else: os.environ[key] = value
if r.source_refs(root) != refs or r.default_player(root) != protected:
error = (error+'; ' if error else '')+'protected root/profile changed'
if checker.input_map(root) != inputs_before:
error = (error+'; ' if error else '')+'spec/fixture/controller inputs changed'
r.write(out/'runner-ended.json', {'error':error, 'cleanup':cleanup, 'cleanup_errors':cleanup_errors,
'phases':phases, 'formal_acceptance':False, 'native_Windows':'pending'})
checker = module('p08_placement_checker', root/'tools/check_p08_normal_placement_evidence.py')
r.write(out/'index.json', {'schema':'p08-normal-placement-v1', 'commit':commit,
'source_map':selected['map'], 'inputs':inputs_before,
'marker_contract_sha256':r.sha(root/CONTRACT),
'spec_sha256':hashlib.sha256((root/'docs/spec/acceptance/P08-03.json').read_text().strip().encode()).hexdigest(),
'cases':cases, 'phases':phases, 'conditions':list(CONDITIONS),
'artifacts':{str(p.relative_to(out)):{'bytes':p.stat().st_size,'sha256':r.sha(p)}
for p in out.rglob('*') if p.is_file() and 'stage' not in p.relative_to(out).parts},
'os_proof':os_proof, 'regression':None,
'acceptance':'pending', 'collection_error':error})
return 1 if error or cleanup_errors else 0
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--project-root', type=absolute, required=True)
parser.add_argument('--output', type=absolute, required=True)
parser.add_argument('--commit', required=True)
parser.add_argument('--godot', default='godot')
parser.add_argument('--input-mode', choices=('handler', 'native_os'), default='handler')
parser.add_argument('--collect', action='store_true', required=True)
return collect(parser.parse_args(argv))
if __name__ == '__main__':
raise SystemExit(main())
tools/test_p08_normal_placement_evidence.py
+259
View File
@@ -0,0 +1,259 @@
#!/usr/bin/env python3
"""Whole-checker positive and semantic corruption tests; synthetic proof only."""
import copy
import importlib.util
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch
from types import SimpleNamespace
import shutil
import contextlib
import io
import os
sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[1]
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
C = module('placement_checker_test', ROOT/'tools/check_p08_normal_placement_evidence.py')
F = module('placement_synthetic_fixture', ROOT/'tests/fixtures/p08-03/checker_fixture.py')
D = module('placement_collector_test', ROOT/'tools/run_p08_normal_placement.py')
class EvidenceTests(unittest.TestCase):
def setUp(self):
scratch = ROOT/'tests/fixtures/p08-03/.scratch'
scratch.mkdir(exist_ok=True)
self.temp = tempfile.TemporaryDirectory(dir=scratch)
self.addCleanup(self.temp.cleanup)
self.root, self.out = F.make(Path(self.temp.name), ROOT, C)
def edit(self, path, callback):
p = self.out/path
data = json.loads(p.read_text())
callback(data)
F.put(p, data)
F.repin(self.out)
def reject(self, text):
errors = C.check(self.root, self.out)
self.assertTrue(errors, 'corrupted proof accepted')
self.assertIn(text, ' '.join(errors))
def test_positive_whole_checker(self):
self.assertEqual([], C.check(self.root, self.out))
def test_verify_only_cli_uses_build_output_directory(self):
output = io.StringIO()
with patch.dict(os.environ, {'BUILD_OUTPUT_DIR':str(self.out.parent)}), contextlib.redirect_stdout(output):
self.assertEqual(0,C.main(['--verify-only','--project-root',str(self.root)]))
self.assertEqual('pending',json.loads(output.getvalue())['acceptance'])
def test_verify_only_cli_rejects_semantic_corruption(self):
self.edit('pair/writer-actions.json', lambda d: d['actions'][0].update(status='配置可能'))
output = io.StringIO()
with patch.dict(os.environ, {'BUILD_OUTPUT_DIR':str(self.out.parent)}), contextlib.redirect_stdout(output):
self.assertEqual(1,C.main(['--verify-only','--project-root',str(self.root)]))
self.assertIn('reason mismatch',json.loads(output.getvalue())['deficits'])
def test_reason_mismatch(self):
self.edit('pair/writer-actions.json', lambda d: d['actions'][0].update(status='配置可能'))
self.reject('reason mismatch')
def test_world_cell_drift(self):
self.edit('pair/writer-actions.json', lambda d: d['actions'][0]['position'].update(selected_cell=[9, 9, 0]))
self.reject('world-cell drift')
def test_transformed_viewport_corruption(self):
self.edit('pair/writer-actions.json', lambda d: d['actions'][0]['position'].update(viewport=[1,2]))
self.reject('transformed viewport mismatch')
def test_stale_source(self):
(self.root/C.TEST).write_text('changed')
self.reject('stale source')
def test_stale_commit(self):
self.edit('index.json', lambda d: d.update(commit='0'*40))
self.reject('stale commit')
def test_stale_input(self):
(self.root/'docs/traceability/original/BASELINE.txt').write_text('changed')
self.reject('stale input')
def test_missing_os(self):
self.edit('index.json', lambda d: d.update(os_proof=None))
self.reject('missing artifact')
def test_fake_os_method(self):
self.edit('os-pair/unsupported-os.json', lambda d: d.update(method='handler'))
self.reject('fake OS provenance')
def test_fake_os_pid(self):
self.edit('os-pair/unsupported-os.json', lambda d: d.update(target_pid=555))
self.reject('fake OS provenance')
def test_same_pid_resume(self):
self.edit('runs/reader/reader-receipt.json', lambda d: d.update(actual_pid=101))
self.reject('duplicate actual PID')
def test_orphan_occupancy(self):
path = 'pair/writer-demolition-after.json'
model = C.decode(json.loads((self.out/path).read_text()))
model['systems']['runtime']['world']['occupancies'].append({'entity_id': 'ghost', 'cells': [[-8, 0, 0]]})
F.put(self.out/path, F.typed(model)); F.repin(self.out)
self.reject('orphan/lost occupancy')
def test_partial_demolition(self):
path = 'pair/writer-demolition-after.json'
model = C.decode(json.loads((self.out/path).read_text()))
model['systems']['runtime']['facilities']['instances'].append({'instance_id': 'f1', 'occupied_cells': [[-8, 0, 0]]})
F.put(self.out/path, F.typed(model)); F.repin(self.out)
self.reject('partial demolition')
def test_refusal_partial_resources(self):
path = 'pair/writer-refused_demolition-after.json'
model = C.decode(json.loads((self.out/path).read_text()))
model['game_state']['resources']['gold'] -= 1
F.put(self.out/path, F.typed(model)); F.repin(self.out)
self.reject('refusal partially applied')
def test_wrong_resume(self):
path = 'pair/reader-checkpoint.json'
model = C.decode(json.loads((self.out/path).read_text()))
model['clock']['absolute_minute'] += 1
F.put(self.out/path, F.typed(model)); F.repin(self.out)
self.reject('wrong resume model')
def test_missing_formal_member(self):
(self.out/'pair/formal/journal.log').unlink()
F.repin(self.out)
self.reject('journal.log')
def test_missing_model_member(self):
path = 'pair/writer-unsupported-before.json'
model = C.decode(json.loads((self.out/path).read_text()))
del model['systems']['runtime']['inventory']
F.put(self.out/path, F.typed(model)); F.repin(self.out)
self.reject('missing whole Runtime members')
def test_missing_clamp_boundary(self):
self.edit('pair/writer-actions.json', lambda d: d['actions'][-1].update(steps=[1]*18))
self.reject('wheel clamp sequence')
def test_no_actual_pan(self):
self.edit('pair/writer-actions.json', lambda d: d['actions'][-1].update(pan_after=[0,0]))
self.reject('move-mode wheel did not pan')
def test_handler_command_must_remain_headless(self):
self.edit('runs/writer/writer-receipt.json', lambda d: d['command'].remove('--headless'))
self.reject('handler command contract')
def test_os_headless_masquerade(self):
self.edit('os-runs/writer/ended.json', lambda d: d['command'].insert(1, '--headless'))
self.reject('OS native argv')
def test_stale_os_recorder(self):
self.edit('os-runs/writer/ended.json', lambda d: d.update(recorder_sha256='0'*64))
self.reject('OS controller/recorder SHA')
def test_os_process_unended(self):
self.edit('os-runs/writer/ended.json', lambda d: d.update(engine_ended=False))
self.reject('OS native failure')
def test_os_default_player_changed(self):
self.edit('os-runs/writer/ended.json', lambda d: d.update(default_after={'changed':True}))
self.reject('OS protected state changed')
def test_os_actual_observation_missing(self):
self.edit('os-pair/unsupported-1/observed.json', lambda d: d.update(events=[]))
self.reject('missing OS events')
def test_os_ack_for_another_request(self):
self.edit('os-pair/unsupported-1/ack.json', lambda d: d.update(request_sha256='0'*64))
self.reject('OS ack request mismatch')
def test_refusal_cannot_be_relabelled(self):
self.edit('pair/writer-actions.json', lambda d: d['actions'][4].update(expected='OTHER_REASON'))
self.reject('required refusal reason changed')
def test_demolition_cannot_be_relabelled(self):
self.edit('pair/writer-actions.json', lambda d: d['actions'][9].update(kind='camera'))
self.reject('operation kind mismatch')
def test_png_signature_is_not_a_screenshot(self):
(self.out/'os-pair/screen.png').write_bytes(b'\x89PNG\r\n\x1a\nFAKE')
F.repin(self.out)
self.reject('invalid OS screenshot')
def test_omitted_os_lifecycle_request(self):
self.edit('os-pair/writer-requests.json', lambda d: d['requests'].pop())
self.reject('omitted OS lifecycle requests')
def test_os_audio_boolean_cannot_replace_owner_evidence(self):
self.edit('os-runs/writer/teardown/owned-audio/owner_1.json', lambda d: d['final_targets'][0].update(alive=True))
self.reject('real WeakRef release')
def collector_mock(self, fail_wait=False):
"""No native engine: Popen, profiles, live engine probes are all mocked."""
output = Path(self.temp.name)/'mock-collection'
stage = output/'stage'; stage.mkdir(parents=True)
shutil.copyfile(self.root/'project.godot',stage/'project.godot')
public = (stage/'project.godot').read_bytes()
(output/'pair').mkdir()
shutil.copytree(self.out/'pair/formal',output/'pair/initial-formal')
F.put(output/'pair/diagnostic-budget.json',{'setup':'synthetic mock'})
refs = C.runner.source_refs(self.root)
protected = {'mock_default_player':True}
captured = []; cleaned = []
def spawn(argv, **kw):
self.assertNotIn('--headless', argv)
self.assertEqual(['--path',str(stage),'--script'],argv[1:4])
role = kw['env']['P08_PLACEMENT_ROLE']
profile_text = (stage/'project.godot').read_text()
name = __import__('re').search(r'config/custom_user_dir_name="([^"]+)"',profile_text).group(1)
pid = 801+len(captured)
process = SimpleNamespace(pid=pid)
def wait(timeout):
self.assertEqual(900,timeout)
if fail_wait:
raise TimeoutError('mock Root acknowledgement timeout')
kw['stdout'].write('P08 passive OS placement: 2 checks, 0 failures\n')
F.put(output/('os-pair/'+role+'-profile.json'),{'pid':pid,'user_data_dir':str(Path(self.temp.name)/name),'display':'mock'})
if role == 'writer': shutil.copytree(output/'pair/initial-formal',output/'os-pair/formal')
return 0
process.wait = wait; captured.append(process)
return process
def cleanup(children):
cleaned.extend(p.pid for p in children)
return ([{'pid':p.pid,'engine_ended':True} for p in children], [])
r = SimpleNamespace(require=C.runner.require, inventory=C.runner.inventory,
sha=C.sha, read=C.load, write=C.runner.write, copy_source=shutil.copyfile,
source_refs=lambda root:refs, default_player=lambda root:protected,
profile_path=lambda name:Path(self.temp.name)/name,
private_project=C.runner.private_project, fresh_environment=C.runner.fresh_environment,
SUPPORT=ROOT/'tools/regression', module=lambda *a:SimpleNamespace(ordinary_gate=lambda *a:{'mock':True}))
outer = SimpleNamespace(native=SimpleNamespace(helper={'live_engines':lambda:[]}),finish_captured=cleanup)
with patch.object(D.subprocess,'Popen',side_effect=spawn), contextlib.redirect_stdout(io.StringIO()):
if fail_wait:
with self.assertRaisesRegex(AssertionError,'native OS collection failed'):
D.collect_os(r,self.root,output,stage,str(Path(self.temp.name)/'synthetic-engine-not-executable'),'1'*40,refs,protected,outer)
else:
phases = D.collect_os(r,self.root,output,stage,str(Path(self.temp.name)/'synthetic-engine-not-executable'),'1'*40,refs,protected,outer)
self.assertEqual(2,len(phases))
self.assertEqual([p.pid for p in captured],cleaned)
self.assertEqual(public,(stage/'project.godot').read_bytes())
self.assertTrue((output/'os-runs/writer/ended.json').exists())
def test_mock_native_controller_two_private_pids(self):
self.collector_mock()
def test_mock_native_timeout_restores_project_and_cleans_owned_child(self):
self.collector_mock(fail_wait=True)
if __name__ == '__main__':
unittest.main()