139 lines
7.5 KiB
Python
139 lines
7.5 KiB
Python
"""Launch the prepared trial only with the exact reviewed e44-to-dd39 reuse attestation."""
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
import argparse
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
TRIAL = ROOT / '.local/native-qa/lifecycle-palette-20260913-01'
|
|
|
|
|
|
def require(condition, reason):
|
|
if not condition:
|
|
raise ValueError(reason)
|
|
|
|
|
|
def identity(data):
|
|
return {'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()}
|
|
|
|
|
|
def verify_reference(item):
|
|
path = (ROOT / item['path']).resolve()
|
|
require(path.is_relative_to(ROOT), 'Provenance path escapes checkout')
|
|
require(identity(path.read_bytes()) == {key: item[key] for key in ('bytes', 'sha256')},
|
|
'Native fixture provenance changed: ' + item['path'])
|
|
return path
|
|
|
|
|
|
def verify_topology():
|
|
path = TRIAL / 'topology-preparation.json'
|
|
topology = json.loads(path.read_bytes())
|
|
require(verify_reference(topology['original_preparation']) == TRIAL / 'preparation.json',
|
|
'Topology must reference the original unchanged trial preparation')
|
|
wanted = {TRIAL / label / name for label in ('topology-baseline', 'topology-work')
|
|
for name in ('project.arcadia', 'rationale.arcadia')}
|
|
copied = []
|
|
for item in topology['files']:
|
|
copied.append(verify_reference(item))
|
|
source = verify_reference(item['source'])
|
|
require(copied[-1].read_bytes() == source.read_bytes(), 'Topology copy differs from source')
|
|
require(len(copied) == 4 and set(copied) == wanted, 'Unexpected topology fixture inventory')
|
|
for item in topology['provenance'] + [topology['oracle'], topology['runtime'], topology['helper']]:
|
|
verify_reference(item)
|
|
require(topology['verified_core_config_input_count'] == 110, 'Unexpected topology core-input scope')
|
|
run_dir = (ROOT / topology['run_directory']).resolve()
|
|
require(run_dir == TRIAL / 'topology-admission-42511f0-r1', 'Unexpected topology admission directory')
|
|
run = json.loads((run_dir / 'run.json').read_bytes())
|
|
require(run['exit'] == 0 and run['inputs_unchanged'] is True, 'Topology source admission did not close cleanly')
|
|
require(run['oracle_sha256'] == topology['oracle']['sha256'], 'Topology oracle identity mismatch')
|
|
require(run['command'] == [str(ROOT / topology['oracle']['path']), 'same',
|
|
str(TRIAL / 'topology-baseline/project.arcadia'), str(TRIAL / 'topology-work/project.arcadia')],
|
|
'Unexpected topology admission command')
|
|
for stream in ('stdout', 'stderr'):
|
|
require(verify_reference(run[stream]) == run_dir / (stream + '.log'), 'Wrong topology admission log')
|
|
scoped = json.loads((run_dir / 'verified-current-core-config-inputs.json').read_bytes())
|
|
require(len(scoped) == 110, 'Unexpected verified topology inputs')
|
|
for item in scoped:
|
|
verify_reference(item)
|
|
return {'topology_preparation_manifest': identity(path.read_bytes()),
|
|
'topology_admission_receipt': identity((run_dir / 'run.json').read_bytes()),
|
|
'topology_admission_scope': topology['scope']}
|
|
|
|
|
|
def checked(stage, source):
|
|
helper = ROOT / '.local/stage-palette-gui-testfix.py'
|
|
spec = importlib.util.spec_from_file_location('palette_exact_testfix_stage', helper)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
require(source == module.SOURCE, 'Only the explicit dd39 acceptance source is allowed')
|
|
record = module.validate_staged(stage)
|
|
require(record['full_test_source'] == module.FULL, 'Full suite must retain its actual e44 source')
|
|
require(record['full_suite_reuse']['policy'] == 'exact-readonly-iterator-testfix-dd39-v1', 'Missing exact reuse policy')
|
|
require(record['full_suite_reuse']['delta'] == module.validate_delta(), 'Reviewed source transformation changed')
|
|
require((TRIAL / 'preparation.json').is_file(), 'Original prepared trial is required')
|
|
for name in ('launch.json', 'closure.json', 'stdout.log', 'stderr.log'):
|
|
require(not (TRIAL / name).exists(), 'Trial launch already attempted: ' + name)
|
|
preparation = json.loads((TRIAL / 'preparation.json').read_bytes())
|
|
for item in preparation['files']:
|
|
path = (TRIAL / item['path']).resolve()
|
|
require(path.is_relative_to(TRIAL.resolve()), 'Unsafe preparation path')
|
|
require(identity(path.read_bytes()) == {key: item[key] for key in ('bytes', 'sha256')},
|
|
'Prepared fixture changed before first launch: ' + item['path'])
|
|
return record, {
|
|
**verify_topology(),
|
|
'stage_verifier': identity(helper.read_bytes()),
|
|
'launcher': identity(Path(__file__).read_bytes()),
|
|
'original_launcher': identity((ROOT / '.local/lifecycle-palette-native-session.py').read_bytes()),
|
|
'stage_manifest': identity((Path(stage) / 'manifest.json').read_bytes()),
|
|
'preparation_manifest': identity((TRIAL / 'preparation.json').read_bytes()),
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('mode', choices=('check', 'launch'))
|
|
parser.add_argument('stage', type=Path)
|
|
parser.add_argument('source')
|
|
args = parser.parse_args()
|
|
stage = args.stage.resolve()
|
|
record, provenance = checked(stage, args.source)
|
|
if args.mode == 'check':
|
|
print(json.dumps({'ready': True, 'source': record['source'], 'full_test_source': record['full_test_source'],
|
|
'trial': str(TRIAL), 'provenance': provenance}, indent=2))
|
|
return
|
|
# The clean private profile and process/exit recording match the original
|
|
# launcher. No original helper, preparation file, or fixture is overwritten.
|
|
env = {key: os.environ[key] for key in ['SystemRoot', 'WINDIR', 'SystemDrive', 'ProgramData',
|
|
'ALLUSERSPROFILE', 'PUBLIC', 'COMSPEC', 'PATHEXT', 'NUMBER_OF_PROCESSORS',
|
|
'PROCESSOR_ARCHITECTURE', 'OS'] if key in os.environ}
|
|
profile = TRIAL / 'profile'
|
|
windows = Path(env['SystemRoot'])
|
|
env.update({'PATH': os.pathsep.join(map(str, [stage, windows / 'System32', windows])),
|
|
'USERPROFILE': str(profile), 'LOCALAPPDATA': str(profile / 'AppData/Local'),
|
|
'APPDATA': str(profile / 'AppData/Roaming'), 'TEMP': str(TRIAL / 'temp'), 'TMP': str(TRIAL / 'temp')})
|
|
with (TRIAL / 'stdout.log').open('xb') as out, (TRIAL / 'stderr.log').open('xb') as err:
|
|
process = subprocess.Popen([str(stage / 'arcadia-studio.exe')], cwd=TRIAL, env=env,
|
|
stdin=subprocess.DEVNULL, stdout=out, stderr=err, creationflags=subprocess.CREATE_NO_WINDOW)
|
|
launch = {'utc': datetime.now(timezone.utc).isoformat(), 'pid': process.pid,
|
|
'source': record['source'], 'full_test_source': record['full_test_source'],
|
|
'full_suite_reuse': record['full_suite_reuse'], 'provenance': provenance,
|
|
'binary': str(stage / 'arcadia-studio.exe'),
|
|
'sha256': record['artifacts']['arcadia-studio.exe']['sha256'], 'private_profile': str(profile)}
|
|
with (TRIAL / 'launch.json').open('x', encoding='utf-8', newline='\n') as stream:
|
|
stream.write(json.dumps(launch, indent=2) + '\n')
|
|
print(json.dumps(launch), flush=True)
|
|
code = process.wait()
|
|
closure = {'utc': datetime.now(timezone.utc).isoformat(), 'pid': process.pid, 'exit': code}
|
|
with (TRIAL / 'closure.json').open('x', encoding='utf-8', newline='\n') as stream:
|
|
stream.write(json.dumps(closure, indent=2) + '\n')
|
|
print(json.dumps(closure), flush=True)
|
|
raise SystemExit(code)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|