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

P08-02: 旧配置の安全な変更

Browse Source
This commit is contained in:
Agent Control Runner
2026-10-13 20:05:01 +09:00
parent e196578c1a
commit 6c7c724410
12 changed files with 18616 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/check_p08_migration_evidence.py
+295
View File
@@ -0,0 +1,295 @@
#!/usr/bin/env python3
"""Read-only artifact integrity gate. Success NEVER accepts original P08-02."""
import argparse
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import re
import sys
sys.dont_write_bytecode = True
HERE = Path(__file__).resolve().parent
CONTRACT = 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json'
SPEC_SHA = '19f02bb7912a4d7354e565186f02366fc1d4dd698468d26142e8204fd14ebfb0'
CONDITIONS = ('mixed_construction','occupancy_change','actual_actor_positions',
'combined_rescue_logistics_ev','no_overlap_orphan_loss','unsolvable_preservation')
MEMBERS = ('world.bin','manifest.json','thumbnail.png','journal.log')
CASES = ('off-positive','off-refusal','on-positive','on-refusal')
LABELS = ['checkpoint'] + ['minute-%03d' % n for n in range(1,161)] + ['events']
def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def load(path): return json.loads(Path(path).read_bytes())
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 decode(node):
"""Only decode for semantic checks; whole typed nodes remain comparison authority."""
t = node['type']
if t == 'Dictionary':
pairs = [(decode(k),decode(v)) for k,v in node['entries']]
keys = [json.dumps(k,sort_keys=True) for k,v in pairs]
if len(keys) != len(set(keys)): raise ValueError('duplicate typed dictionary key')
return {k:v for k,v in pairs}
if t == 'Array': return [decode(n) for n in node['items']]
text = node['text']
if t in ('String','StringName'): return json.loads(text.removeprefix('&'))
if t == 'int': return int(text)
if t == 'float': return float(text)
if t == 'bool':
if text not in ('true','false'): raise ValueError('invalid bool')
return text == 'true'
if t == 'Nil': return None
if t in ('Vector2i','Vector3i','Vector2','Vector3'):
return [float(x.strip()) if '.' in x else int(x.strip()) for x in text.strip('()').split(',')]
# Preserve unfamiliar scalar types verbatim; never silently drop them.
return (t,text)
def indexed(rows, key):
result = {row[key]:row for row in rows}
if len(result) != len(rows): raise ValueError('duplicate '+key)
return result
def semantics(snapshot, identity, initial=False):
"""Independent ID/cell/reference checks on actual serialized Runtime state."""
r = snapshot['systems']['runtime']
guests = indexed(r['guests']['guests'],'guest_id')
staff = indexed(r['staff']['staff'],'staff_id')
if set(guests) & set(staff): raise ValueError('duplicate actor across domains')
jobs = indexed(r['construction']['jobs'],'job_id')
facilities = indexed(r['construction']['facilities'],'instance_id')
operational = indexed(r['facilities']['instances'],'instance_id')
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')})
occupied = indexed(r['world']['occupancies'],'entity_id')
cells = set()
for owner,row in occupied.items():
actual = {tuple(c) for c in row['cells']}
if len(actual) != len(row['cells']) or cells & actual: raise ValueError('overlap occupancy')
cells |= actual
if owner not in owners or actual != {tuple(c) for c in owners[owner]}: raise ValueError('orphan occupancy/ownership')
if set(owners) != set(occupied): raise ValueError('lost occupancy owner')
for fid,facility in operational.items():
if fid not in facilities or facility['occupied_cells'] != facilities[fid]['occupied_cells']:
raise ValueError('operational/construction ownership mismatch')
for jid in identity['jobs']:
if jid not in jobs: raise ValueError('lost paid job')
if initial and (jobs[jid]['progress'] <= 0 or jobs[jid]['state'] != 'BUILDING' or len(jobs[jid]['reserved_cells']) != 6):
raise ValueError('missing paid old-size progress/reservation')
if len(identity['jobs']) < 2 or identity['dorm'] not in facilities: raise ValueError('mixed jobs/completed facility missing')
tasks = indexed(r['staff']['tasks'],'task_id')
if 'patient' not in staff or 'carrier' not in staff: raise ValueError('lost patient/carrier')
rescue = tasks[identity['rescue_task']]
if rescue['target_entity_id'] != 'patient' or rescue['rescue_facility_id'] not in facilities:
raise ValueError('rescue patient/destination reference')
if initial and (rescue['state'] != 'MOVING' or rescue['assigned_staff_id'] != 'carrier' or staff['patient']['health'] != 0):
raise ValueError('rescue is not actually in flight')
inventories = indexed(r['inventory']['inventories'],'owner_inventory_id')
requests = indexed(r['logistics_requests'],'request_id')
reservations = indexed(r['inventory']['reservations'],'reservation_id')
if not identity['logistics_requests']: raise ValueError('missing original food request')
for rid in identity['logistics_requests']:
req = requests[rid]
if req['source_inventory_id'] not in inventories or req['destination_inventory_id'] not in inventories:
raise ValueError('orphan cargo endpoint')
if initial:
carrier = staff[req['assigned_carrier_id']]
cargo = inventories[carrier['carried_inventory_id']]
if not cargo['stacks'] or not req['reservation_ids'] or any(x not in reservations for x in req['reservation_ids']):
raise ValueError('lost cargo/reservation')
trips = indexed(r['elevators']['trips'],'request_id')
actors = set(guests) | set(staff)
for trip in trips.values():
actor = trip['actor_id']
if actor not in actors and actor.removeprefix('guest:').removeprefix('staff:') not in actors:
raise ValueError('orphan EV actor')
if initial:
if len(identity['ev_requests']) != 5 or not set(identity['ev_requests']) <= set(trips): raise ValueError('lost EV request')
if sum(len(c['boarded']) for c in r['elevators']['shafts']) != 4: raise ValueError('EV passengers/queue absent')
if 'p08.walker' not in guests or not r['movement_credits']: raise ValueError('mid-route guest/credit absent')
return r
def typed_differences(a, b, path=()):
if a == b: return []
if a.get('type') != b.get('type'): return [path]
if a['type'] == 'Dictionary':
aa = {str(decode(k)):(k,v) for k,v in a['entries']}; bb = {str(decode(k)):(k,v) for k,v in b['entries']}
if aa.keys() != bb.keys(): return [path]
differences = []
for k in aa:
if aa[k][0] != bb[k][0]: differences.append(path+(k,'@key-type'))
differences += typed_differences(aa[k][1],bb[k][1],path+(k,))
return differences
if a['type'] == 'Array' and a.get('element_type') == b.get('element_type') and a.get('element_class') == b.get('element_class'):
if len(a['items']) != len(b['items']): return [path]
return [p for i,(x,y) in enumerate(zip(a['items'],b['items'])) for p in typed_differences(x,y,path+(str(i),))]
return [path]
# Explicit geometry changes only. No actor, economy, request, event or route wildcard.
# This fixture deliberately places actors away from removed cells: NO route changes approved.
GEOMETRY = [
r'systems/runtime/world/occupancies/\d+/cells(?:/.*)?',
r'systems/runtime/world/dirty_cells(?:/.*)?',
r'systems/runtime/world/revisions/(world|occupancy)',
r'systems/runtime/construction/jobs/\d+/reserved_cells(?:/.*)?',
r'systems/runtime/construction/facilities/\d+/occupied_cells(?:/.*)?',
r'systems/runtime/facilities/instances/\d+/occupied_cells(?:/.*)?',
]
def check(project_root, evidence_dir):
root, evidence = Path(project_root).resolve(), Path(evidence_dir).resolve()
deficits = []
def need(ok, reason):
if not ok: deficits.append(reason)
def guarded(label, fn):
try: return fn()
except (KeyError,ValueError,TypeError,OSError,IndexError,AttributeError) as exc:
deficits.append(label+': '+str(exc)); return None
index = guarded('index',lambda:load(evidence/'index.json'))
if not isinstance(index,dict): return deficits or ['index is not an object']
def artifact(name):
if not isinstance(name,str) or Path(name).is_absolute() or '..' in Path(name).parts: raise ValueError('unsafe artifact path')
path = evidence/name
if not path.is_file() or any(p.is_symlink() for p in [path,*path.parents] if p != evidence.parent): raise ValueError('missing/aliased artifact '+name)
pin = index['artifacts'][name]
if pin != {'bytes':path.stat().st_size,'sha256':sha(path)}: raise ValueError('artifact hash '+name)
return path
def document(name): return load(artifact(name))
need(index.get('schema') == 'p08-evidence-v1','schema')
need(index.get('acceptance') == 'pending','integrity is not unit acceptance')
need(index.get('collection_error') is None,'collection failure')
def source_gate():
runner = module('p08_readonly_source',root/'tools/run_all_tests.py')
actual = {n:p['sha256'] for n,p in runner.source_refs(root).items()}
need(index['source_map'] == actual,'wrong/stale/incomplete final source map')
spec = load(root/'docs/spec/acceptance/P08-02.json')
need(index['spec_sha256'] == SPEC_SHA == hashlib.sha256((root/'docs/spec/acceptance/P08-02.json').read_text().strip().encode()).hexdigest(),'frozen spec SHA')
for name,pin in spec['current_source_pins'].items(): need(actual.get(name) == pin,'frozen production source '+name)
inputs = {str(p.relative_to(root)):sha(p) for p in (root/'docs/spec/inputs/p08-02').iterdir() if p.is_file()}
need(index['inputs'] == inputs,'frozen input map')
need(index['marker_contract_sha256'] == sha(root/CONTRACT),'marker contract source')
need(re.fullmatch('[0-9a-f]{40}',index['commit']) is not None,'submitted commit identity')
guarded('source',source_gate)
for name in index.get('artifacts',{}): guarded('artifact '+name,lambda name=name:artifact(name))
markers = module('p08_evidence_markers',root/'tools/strict_test_markers.py')
contracts = markers.load_contracts(root/CONTRACT)
pids = set()
def phase(name, role):
receipt = document(name); pid = receipt['actual_pid']
need(type(pid) is int and pid > 0 and pid not in pids,'distinct native PID '+name); pids.add(pid)
parent = str(Path(name).parent)
prefix = '' if parent == '.' else parent+'/'
folder = prefix+receipt['phase']+'/'
start,end,outer = [document(folder+n) for n in ('started.json','ended.json','outer-owned-finally.json')]
for row in (start,end):
need(row['actual_pid'] == pid and row['command'] == receipt['command'],'raw lifecycle PID/argv '+name)
need(end.get('exit_code') == 0 and end.get('engine_ended') is True,'ended exit '+name)
need(start.get('live_identity_exact') is True or start.get('tracked_child_ended_before_initial_probe') is True,'native Popen identity '+name)
need(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 '+name)
native = receipt['native_process']
need(native['exit_code'] == 0 and native['helper_exit_code'] == 0 and native['timed_out'] is False and native['lifecycle_error'] is None,'native exit/errors '+name)
need(receipt['exit_code'] == 0 and receipt['engine_ended'] is True,'phase ended '+name)
if role:
test = 'tests/fixtures/p08-02/'+role+'.gd'
need(receipt['test'] == test and receipt['source_sha256'] == index['source_map'][test],'phase source '+name)
raw = artifact(prefix+receipt['log']).read_text()
parsed = markers.parse_raw(test,raw,{test:{'template':'P08 mixed '+role+': <checks> checks, <failures> failures'}},receipt['source_sha256'])
need(not parsed['errors'],'strict completion '+name+': '+str(parsed['errors']))
profile = receipt['actual_private_profile']['report']
need(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'],'guarded fresh profile '+name)
return receipt
def case_gate(case):
name = case['id']; prefix = case['directory']+'/'
need(name == case['operations']+'-'+case['kind'],'case identity '+name)
need(len(case['phases']) == 2,'writer and reader phases '+name)
receipts = [phase(n,r) for n,r in zip(case['phases'],('writer','reader'))]
def typed(name):
artifact(prefix+name+'.var')
return document(prefix+name+'.json')
identity = decode(typed('identity'))
original = typed('original'); after_live = typed('after-live')
need(original == after_live,'live partial batch '+name)
before_members,after_members = [decode(typed(n)) for n in ('before-members','after-members')]
for package in ('original','result','backup'):
for member in MEMBERS: artifact(prefix+package+'/formal/'+member)
for member in MEMBERS:
need(before_members['/'+member] == sha(artifact(prefix+'original/formal/'+member)),'original overwritten '+name+'/'+member)
need(after_members['/'+member] == sha(artifact(prefix+'result/formal/'+member)),'result member '+name+'/'+member)
need(after_members['.pre_migration.bak/'+member] == sha(artifact(prefix+'backup/formal/'+member)),'backup overwritten '+name+'/'+member)
refusal = case['kind'] == 'refusal'
if refusal: need(before_members == after_members,'refusal original/backup overwritten '+name)
source = semantics(decode(original),identity,True)
need(source['operations']['enabled'] == (case['operations']=='on'),'missing actual Operations mode '+name)
for role,receipt in zip(('writer','reader'),receipts):
proc = decode(typed(role+'-process'))
need(proc['pid'] == receipt['actual_pid'] and proc['failures'] == 0 and proc['operations'] == (case['operations']=='on') and proc['refusal'] == refusal,'fixture/native binding '+name+'/'+role)
checkpoint = typed('writer-checkpoint')
if refusal: need(original == checkpoint,'partial earlier candidate '+name)
else:
for path in typed_differences(original,checkpoint):
need(any(re.fullmatch(pattern,'/'.join(path)) for pattern in GEOMETRY),'unapproved typed migration delta '+name+'/'+ '/'.join(path))
migrated = decode(checkpoint)['systems']['runtime']
jobs = indexed(migrated['construction']['jobs'],'job_id')
need(all(len(jobs[j]['reserved_cells']) == 4 for j in identity['jobs']),'positive batch not applied '+name)
for label in LABELS:
writer,reader = typed('writer-'+label),typed('reader-'+label)
need(writer == reader,'cross-PID full typed state/events '+name+'/'+label)
need(sha(artifact(prefix+'writer-'+label+'.var')) == sha(artifact(prefix+'reader-'+label+'.var')),'cross-PID raw Variant '+name+'/'+label)
if label != 'events': semantics(decode(writer),identity)
final = decode(typed('writer-minute-160'))['systems']['runtime']
need(indexed(final['staff']['staff'],'staff_id')['patient']['health'] == 100,'patient lost/untreated '+name)
final_requests = indexed(final['logistics_requests'],'request_id')
need(all(final_requests[i]['state']=='COMPLETED' for i in identity['logistics_requests']),'original cargo completion '+name)
events = decode(typed('writer-events'))
for rid in identity['ev_requests']:
need(sum(e['event_type']=='ELEVATOR_ARRIVED' and e['payload'].get('request_id')==rid for e in events)==1,'EV original request exactly once '+name+'/'+rid)
cases = index.get('cases',[])
need([c.get('id') for c in cases] == list(CASES),'missing/duplicate ON/OFF positive/refusal case')
for case in cases: guarded('case '+str(case.get('id')),lambda case=case:case_gate(case))
for condition in CONDITIONS:
expected = [c for c in CASES if condition != 'unsolvable_preservation' or c.endswith('refusal')]
need(index.get('conditions',{}).get(condition) == expected,'condition mapping '+condition)
def os_gate():
proof = document(index['os_proof'])
need(proof['commit'] == index['commit'] and proof['source_map'] == index['source_map'],'OS current candidate')
need(proof['input_method'] == 'native_os' and proof['setup'] == 'controlled_old_save','OS input provenance/setup')
actions = proof['actions']
need([a['action'] for a in actions] == ['preview','cancel','confirm','changed_source_rejection'],'OS actions missing')
for action in actions:
raw = document(action['input_receipt'])
need(raw['method'] == 'native_os' and raw['event_type'] in ('mouse','keyboard') and type(raw['target_pid']) is int and raw['target_pid'] > 0 and raw['timestamp_ns'] > 0,'fake OS signal')
for key in ('screen_before','screen_after','model_before','model_after','formal_before','formal_after'): artifact(action[key])
if action['action'] in ('preview','cancel','changed_source_rejection'):
need(document(action['formal_before']) == document(action['formal_after']),'OS formal preservation '+action['action'])
guarded('OS proof',os_gate)
def regression_gate():
proof = document(index['regression']); run_input = document(proof['input']); ended = document(proof['ended'])
need(run_input['selected_source_map'] == index['source_map'],'final regression source map')
need(proof['commit'] == index['commit'],'final regression commit')
need(not ended['error'] and not ended['cleanup_errors'] and not ended['guard_errors'],'regression failure')
runner = module('p08_regression_roster',root/'tools/run_all_tests.py')
roster = runner.roster(root)
need(run_input['roster'] == roster,'full final roster')
rows = [x for x in ended['phases'] if x.get('test')]
need([x['test'] for x in rows] == roster,'full final regression receipt')
for row in rows:
raw = artifact(proof['directory']+'/'+row['log']).read_text()
need(not markers.parse_raw(row['test'],raw,contracts,index['source_map'][row['test']])['errors'],'regression strict '+row['test'])
need(row['exit_code'] == 0 and row['engine_ended'] is True,'regression native exit '+row['test'])
guarded('final regression',regression_gate)
return sorted(set(deficits))
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')
deficits = ['BUILD_OUTPUT_DIR missing'] if not build else check(args.project_root,Path(build)/'p08-evidence')
print(json.dumps({'integrity':'deficient' if deficits else 'verified','deficits':deficits,
'acceptance':'pending','conditions':{c:'pending' for c in CONDITIONS}},ensure_ascii=False,indent=2))
return 1 if deficits else 0
if __name__ == '__main__': raise SystemExit(main())
tools/run_p08_mixed_migration.py
+141
View File
@@ -0,0 +1,141 @@
#!/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
sys.dont_write_bytecode = True
HERE = Path(__file__).resolve().parent
CONTRACT = 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json'
CONDITIONS = ('mixed_construction', 'occupancy_change', 'actual_actor_positions',
'combined_rescue_logistics_ev', 'no_overlap_orphan_loss', 'unsolvable_preservation')
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
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 out.exists(), 'fresh output; never overwrite earlier proof')
refs = r.source_refs(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)
for role in ('writer','reader'):
name = 'tests/fixtures/p08-02/'+role+'.gd'
contracts[name] = {'test':name, 'template':'P08 mixed '+role+': <checks> checks, <failures> failures', 'required':True}
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-mixed-v1', 'commit':commit,
'selected_source_map':selected['map'], 'engine':engine, 'engine_sha256':r.sha(engine),
'owner_pid':os.getpid(), 'default_player':protected, 'diagnostic_seed':42588})
phases = []; previous = {}; cases = []; error = None
deadline = time.monotonic()+900
saved_env = {k:os.environ.get(k) for k in ('P08_PAIR_ROOT','P08_OPERATIONS','P08_CASE')}
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)
for operations in ('off','on'):
for case in ('positive','refusal'):
name = operations+'-'+case; pair = out/'pairs'/name
os.environ.update(P08_PAIR_ROOT=str(pair), P08_OPERATIONS=operations, P08_CASE=case)
case_phases = []
preserved = {}
for role in ('writer','reader'):
label = name+'-'+role
test = 'tests/fixtures/p08-02/'+role+'.gd'
phase(label, test)
actual = r.inventory(pair)
r.require(all(actual.get(n) == pin for n,pin in preserved.items()), 'reader preserved every writer byte')
preserved = actual
case_phases.append(phases[-1])
cases.append({'id':name, 'operations':operations, 'kind':case,
'directory':'pairs/'+name, 'phases':case_phases, 'files':preserved})
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
r.require(r.source_refs(root) == refs and r.default_player(root) == protected, 'protected root/profile unchanged')
r.write(out/'runner-ended.json', {'error':error, 'cleanup':cleanup, 'cleanup_errors':cleanup_errors,
'phases':phases, 'formal_acceptance':False, 'native_Windows':'pending'})
inputs = {str(p.relative_to(root)):r.sha(p) for p in (root/'docs/spec/inputs/p08-02').iterdir() if p.is_file()}
# The collector is intentionally incomplete until Root supplies OS and regression artifacts.
r.write(out/'index.json', {'schema':'p08-evidence-v1', 'commit':commit,
'source_map':selected['map'], 'inputs':inputs, 'marker_contract_sha256':r.sha(root/CONTRACT),
'spec_sha256':hashlib.sha256((root/'docs/spec/acceptance/P08-02.json').read_text().strip().encode()).hexdigest(),
'cases':cases, 'phases':phases, 'conditions':{c:[x['id'] for x in cases if c != 'unsolvable_preservation' or x['kind']=='refusal'] for c in 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':None, '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('--collect', action='store_true', required=True)
return collect(parser.parse_args(argv))
if __name__ == '__main__':
raise SystemExit(main())
tools/test_p08_migration_evidence.py
+187
View File
@@ -0,0 +1,187 @@
#!/usr/bin/env python3
"""Synthetic evidence ONLY: disposable end-to-end check() positives and corruptions.
No Godot/OS execution, no actual acceptance proof. Mutations update hashes so the
semantic gate, not merely a stale digest, must reject invalid state.
"""
import copy
import hashlib
import json
from pathlib import Path
import tempfile
import unittest
import check_p08_migration_evidence as c
ROOT = Path(__file__).resolve().parent.parent
BANNER = 'Godot Engine v4.7.2.stable.official.ed1daf0bf - https://godotengine.org\n'
def typed(value):
if isinstance(value,dict): return {'type':'Dictionary','entries':[[typed(k),typed(v)] for k,v in value.items()]}
if isinstance(value,list): return {'type':'Array','element_type':0,'element_class':'','items':[typed(x) for x in value]}
kind = 'Nil' if value is None else 'bool' if type(value) is bool else 'int' if type(value) is int else 'String'
return {'type':kind,'text':json.dumps(value)}
class Proof:
def __init__(self, path):
self.path = path
runner = c.module('p08_synthetic_source',ROOT/'tools/run_all_tests.py')
self.runner = runner
source = {n:p['sha256'] for n,p in runner.source_refs(ROOT).items()}
self.index = {'schema':'p08-evidence-v1','commit':'e196578c1a4ad2fdf84642565bd102e524be386f',
'source_map':source,'spec_sha256':c.SPEC_SHA,'marker_contract_sha256':c.sha(ROOT/c.CONTRACT),
'inputs':{str(p.relative_to(ROOT)):c.sha(p) for p in (ROOT/'docs/spec/inputs/p08-02').iterdir() if p.is_file()},
'cases':[],'conditions':{k:[x for x in c.CASES if k!='unsolvable_preservation' or x.endswith('refusal')] for k in c.CONDITIONS},
'artifacts':{},'acceptance':'pending','collection_error':None,'phases':[],
'os_proof':'os.json','regression':'regression.json'}
for number,name in enumerate(c.CASES): self.case(name,number)
self.os(); self.regression(); self.save()
def put(self,name,value):
path = self.path/name; path.parent.mkdir(parents=True,exist_ok=True)
path.write_bytes(value if isinstance(value,bytes) else json.dumps(value).encode())
self.index['artifacts'][name] = {'bytes':path.stat().st_size,'sha256':c.sha(path)}
def variant(self,prefix,name,value):
node = typed(value)
self.put(prefix+name+'.json',node)
# Deliberately not a native Variant: this is an isolated Python proof fixture.
self.put(prefix+name+'.var',json.dumps(node).encode())
def save(self): (self.path/'index.json').write_text(json.dumps(self.index))
def state(self,operations):
jobs = [{'job_id':j,'state':'BUILDING','progress':466,'reserved_cells':[[x,2,0] for x in range(offset,offset+6)],'construction_cost':{'gold':1}}
for j,offset in [('j1',0),('j2',10)]]
facilities = [{'instance_id':'dorm','occupied_cells':[[30,0,1]]},{'instance_id':'clinic','occupied_cells':[[31,0,1]]}]
guests = [{'guest_id':g} for g in ['p08.walker']+['g'+str(n) for n in range(5)]]
staff = [{'staff_id':'patient','health':0},{'staff_id':'carrier'},{'staff_id':'staff_gina','carried_inventory_id':'cargo'}]
r = {'operations':{'enabled':operations},'guests':{'guests':guests},'staff':{'staff':staff,'tasks':[
{'task_id':'rescue','target_entity_id':'patient','rescue_facility_id':'clinic','state':'MOVING','assigned_staff_id':'carrier'}]},
'construction':{'jobs':jobs,'facilities':facilities},'facilities':{'instances':copy.deepcopy(facilities)},
'world':{'occupancies':[{'entity_id':j['job_id'],'cells':copy.deepcopy(j['reserved_cells'])} for j in jobs]+
[{'entity_id':f['instance_id'],'cells':copy.deepcopy(f['occupied_cells'])} for f in facilities]},
'inventory':{'inventories':[{'owner_inventory_id':x,'stacks':[{'item_id':'ingredient/raw_food','quantity':4}]} for x in ('source','dest','cargo')],
'reservations':[{'reservation_id':'reserve'}]},
'logistics_requests':[{'request_id':'food','source_inventory_id':'source','destination_inventory_id':'dest','assigned_carrier_id':'staff_gina','reservation_ids':['reserve'],'state':'IN_TRANSIT'}],
'movement_credits':[{'credit':1000}],
'elevators':{'trips':[{'request_id':'ev'+str(n),'actor_id':'g'+str(n)} for n in range(5)],'shafts':[{'boarded':['g0','g1','g2','g3']}]}}
return {'systems':{'runtime':r}}
def phase(self,name,role,pid):
folder = 'runs/'+name+'/'; test = 'tests/fixtures/p08-02/'+role+'.gd'
command = ['synthetic-native','--script',test]
start = {'actual_pid':pid,'command':command,'live_identity_exact':True,'event':'ENGINE_STARTED','owner_pid':1,'monotonic_ns':1}
end = dict(start,event='ENGINE_ENDED',monotonic_ns=2,exit_code=0,engine_ended=True)
self.put(folder+name+'/started.json',start); self.put(folder+name+'/ended.json',end)
self.put(folder+name+'/outer-owned-finally.json',{'actual_captured_pids':[pid],'all_captured_ended':True,'outer_cleanup_errors':[],'foreign_process_or_group_signals':0})
self.put(folder+'raw.log',(BANNER+'P08 mixed '+role+': 1 checks, 0 failures\n').encode())
receipt = {'phase':name,'actual_pid':pid,'command':command,'exit_code':0,'engine_ended':True,'test':test,
'source_sha256':self.index['source_map'][test],'log':'raw.log',
'native_process':{'exit_code':0,'helper_exit_code':0,'timed_out':False,'lifecycle_error':None},
'actual_private_profile':{'report':{'fresh_before_parent_initialize':True,'profile_writes_before_guard':0,'actual_user_data_dir':'diagnostic-private-'+name,'expected_user_data_dir':'diagnostic-private-'+name}}}
self.put(folder+name+'-receipt.json',receipt)
return folder+name+'-receipt.json'
def case(self,name,number):
operations = name.startswith('on-'); refusal = name.endswith('refusal'); prefix = 'pairs/'+name+'/'
phases = [self.phase(name+'-'+role,role,100+2*number+i) for i,role in enumerate(('writer','reader'))]
self.index['cases'].append({'id':name,'operations':'on' if operations else 'off','kind':'refusal' if refusal else 'positive','directory':prefix[:-1],'phases':phases})
self.index['phases'] += phases
identity = {'jobs':['j1','j2'],'dorm':'dorm','rescue_task':'rescue','logistics_requests':['food'],'ev_requests':['ev'+str(n) for n in range(5)]}
original = self.state(operations)
self.variant(prefix,'identity',identity); self.variant(prefix,'original',original); self.variant(prefix,'after-live',original)
before = {}; after = {}
for member in c.MEMBERS:
raw = ('synthetic '+member).encode()
for package in ('original','result','backup'): self.put(prefix+package+'/formal/'+member,raw)
before['/'+member] = after['/'+member] = hashlib.sha256(raw).hexdigest()
before['.pre_migration.bak/'+member] = after['.pre_migration.bak/'+member] = hashlib.sha256(raw).hexdigest()
self.variant(prefix,'before-members',before); self.variant(prefix,'after-members',after)
checkpoint = copy.deepcopy(original)
if not refusal:
r = checkpoint['systems']['runtime']
for job in r['construction']['jobs']: job['reserved_cells'] = job['reserved_cells'][:4]
for row in r['world']['occupancies'][:2]: row['cells'] = row['cells'][:4]
events = [{'event_type':'ELEVATOR_ARRIVED','payload':{'request_id':rid}} for rid in identity['ev_requests']]
for i,role in enumerate(('writer','reader')):
self.variant(prefix,role+'-process',{'pid':100+2*number+i,'failures':0,'checks':1,'operations':operations,'refusal':refusal})
for label in c.LABELS:
state = copy.deepcopy(checkpoint)
if label == 'minute-160':
state['systems']['runtime']['staff']['staff'][0]['health'] = 100
state['systems']['runtime']['logistics_requests'][0]['state'] = 'COMPLETED'
self.variant(prefix,role+'-'+label,events if label=='events' else state)
def os(self):
proof = {'commit':self.index['commit'],'source_map':self.index['source_map'],'input_method':'native_os','setup':'controlled_old_save','actions':[]}
for name in ('preview','cancel','confirm','changed_source_rejection'):
row = {'action':name,'input_receipt':'os/'+name+'/input.json'}
self.put(row['input_receipt'],{'method':'native_os','event_type':'mouse','target_pid':500,'timestamp_ns':1})
for field in ('screen_before','screen_after','model_before','model_after','formal_before','formal_after'):
row[field] = 'os/'+name+'/'+field+'.json'; self.put(row[field],{'synthetic':'diagnostic'})
proof['actions'].append(row)
self.put('os.json',proof)
def regression(self):
contracts = c.module('p08_synthetic_markers',ROOT/'tools/strict_test_markers.py').load_contracts(ROOT/c.CONTRACT)
roster = self.runner.roster(ROOT); rows = []
for i,test in enumerate(roster):
contract = contracts[test]
def marker(template):
for key in ('checks','passed'): template = template.replace('<'+key+'>','1')
for key in ('failures','failed','skipped'): template = template.replace('<'+key+'>','0')
return template+'\n'
raw = BANNER+marker(contract['template'])
for auxiliary in contract.get('auxiliary',[]): raw += marker(auxiliary['template'])
if contract.get('json_report'):
raw += contract['json_report']['prefix']+json.dumps({'check_summary':{'total':1,'passed':1,'failed':0,'skipped':0},'status':'PASS','runtime_errors':[],'unmet_fixture_fields':[]})+'\n'
log = str(i)+'.log'; self.put('regression/'+log,raw.encode())
rows.append({'test':test,'log':log,'exit_code':0,'engine_ended':True})
self.put('regression/input.json',{'selected_source_map':self.index['source_map'],'roster':roster})
self.put('regression/ended.json',{'error':None,'cleanup_errors':[],'guard_errors':[],'phases':rows})
self.put('regression.json',{'commit':self.index['commit'],'directory':'regression','input':'regression/input.json','ended':'regression/ended.json'})
class EvidenceTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.temp = tempfile.TemporaryDirectory(prefix='p08-checker-')
cls.proof = Proof(Path(cls.temp.name))
cls.baseline = copy.deepcopy(cls.proof.index)
@classmethod
def tearDownClass(cls): cls.temp.cleanup()
def test_positive_reaches_whole_check(self):
self.assertEqual(c.check(ROOT,self.proof.path),[])
def test_corruptions_cannot_be_accepted(self):
p = self.proof
mutations = {
'duplicate_actor':lambda r:r['guests']['guests'].append(copy.deepcopy(r['guests']['guests'][0])),
'orphan_occupancy':lambda r:r['world']['occupancies'][0].update(entity_id='orphan'),
'lost_patient':lambda r:r['staff']['staff'].pop(0),
'lost_cargo':lambda r:r['inventory']['inventories'].pop(),
'lost_ev':lambda r:r['elevators']['trips'].pop(),
}
path = 'pairs/off-positive/original.json'
original = (p.path/path).read_bytes()
for label,mutate in mutations.items():
with self.subTest(label=label):
node = c.decode(json.loads(original)); mutate(node['systems']['runtime'])
p.put(path,typed(node)); p.save()
self.assertTrue(c.check(ROOT,p.path),label)
p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save()
paths = [('partial_batch','pairs/off-refusal/after-live.json'),('unequal_cross_pid','pairs/on-positive/reader-checkpoint.json'),
('overwritten_original','pairs/off-refusal/original/formal/world.bin'),('overwritten_backup','pairs/off-refusal/backup/formal/world.bin')]
for label,path in paths:
with self.subTest(label=label):
original = (p.path/path).read_bytes()
p.put(path,b'{}'); p.save(); self.assertTrue(c.check(ROOT,p.path),label)
p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save()
for label in ('missing_on','wrong_source','stale_source','fake_os_signal','missing_phase','duplicate_pid'):
with self.subTest(label=label):
changed = None
if label=='missing_on': p.index['cases'] = p.index['cases'][:2]
elif label=='wrong_source': p.index['source_map']['autoload/simulation_runtime.gd'] = '0'*64
elif label=='stale_source': del p.index['source_map']['tools/check_p08_migration_evidence.py']
elif label=='missing_phase': p.index['cases'][0]['phases'].pop()
elif label=='fake_os_signal':
changed = 'os/preview/input.json'; original = (p.path/changed).read_bytes()
row = json.loads(original); row['method'] = 'signal_emitted'; p.put(changed,row)
else:
changed = p.index['cases'][0]['phases'][1]; original = (p.path/changed).read_bytes()
row = json.loads(original); row['actual_pid'] = 100; p.put(changed,row)
p.save(); self.assertTrue(c.check(ROOT,p.path),label)
if changed: p.put(changed,original)
p.index = copy.deepcopy(self.baseline); p.save()
if __name__ == '__main__': unittest.main()