Files
ARCADIA-Studio/harness/experiments/cli-packaging-20260911/reviewer.py
T

257 lines
18 KiB
Python

"""Independent CLI packaging review: read-only source checks and isolated CLI runs."""
from pathlib import Path, PurePosixPath
import ast
import hashlib
import json
import os
import re
import shutil
import stat
import subprocess
import sys
import time
import tomllib
import traceback
import uuid
import zipfile
OUT = Path(__file__).resolve().parent
ROOT = OUT.parents[1]
SOURCE = ROOT / '.local/cli-packaging-20260911/source'
NAME = 'ArcadiaCLI-0.2.0-preview.authoring.20260911-x86_64-pc-windows-gnullvm'
ZIP = SOURCE.parent / 'packages' / (NAME + '.zip')
PACKAGE = OUT / 'extracted' / NAME
BASE = '01b76a1e89e93b7992f39e63560aa8b2ca2f69e0'
ASSEMBLY = 'a0a48e82bd43be181caca1fe2c1514c21f2bc405'
ZIP_SHA = '2b051f8ce1ea4f59ee7e233e1fbd516d423aadbfbca5d549755cbe78fcde1af6'
EXE_SHA = 'bad169f8be78dcfa74b32714569c886dd2132727080684fc2ba618d5822e3559'
DLL_SHA = 'f45604a12e8bcd839fd86f667c1be1cc7c4f9da2b3c69ff381ac708f0e1ae7d4'
RESULT = {'started_utc': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'checks': []}
def check(ok, label):
if not ok:
raise AssertionError(label)
RESULT['checks'].append(label)
def sha(b):
return hashlib.sha256(b).hexdigest()
def read_json(p):
return json.loads(p.read_text(encoding='utf-8-sig'))
def record(path, obj):
with path.open('x', encoding='utf-8', newline='\n') as f:
json.dump(obj, f, indent=2, ensure_ascii=False)
f.write('\n')
def snapshot(directory):
files = {}
for p in directory.rglob('*'):
if p.is_symlink() or getattr(p.lstat(), 'st_file_attributes', 0) & 0x400:
raise AssertionError('unexpected reparse path ' + str(p))
if p.is_file():
files[p.relative_to(directory).as_posix()] = p.read_bytes()
return files
def git(path, ref=BASE):
return subprocess.run(['git', 'show', ref + ':' + path], cwd=ROOT, check=True, capture_output=True).stdout
def source_checks():
old = git('scripts/package-artifacts.py').decode()
new = (ROOT / 'scripts/package-artifacts.py').read_text()
original_functions = {f.name: ast.dump(f, include_attributes=False) for f in ast.parse(old).body if isinstance(f, ast.FunctionDef)}
new_functions = {f.name: ast.dump(f, include_attributes=False) for f in ast.parse(new).body if isinstance(f, ast.FunctionDef)}
check(all(new_functions[k] == v for k, v in original_functions.items() if k not in ('document_inputs', 'main')), 'all original ELF/tar/document/notice helpers unchanged outside product selection')
before, after = {'__name__': 'review_baseline'}, {'__name__': 'review_current'}
exec(compile(old, '<baseline packaging>', 'exec'), before)
exec(compile(new, '<current packaging>', 'exec'), after)
counts = {}
for platform in ('windows', 'linux'):
expected = before['document_inputs'](SOURCE, platform)
default = after['document_inputs'](SOURCE, platform)
explicit = after['document_inputs'](SOURCE, platform, 'studio')
check(default == explicit == expected, platform + ' actual full Studio document closure unchanged')
counts[platform] = len(default['files'])
old_notice = git('scripts/windows-notices.ps1').decode().splitlines()
new_notice = (ROOT / 'scripts/windows-notices.ps1').read_text().splitlines()
check(len(old_notice) == len(new_notice) and [i for i, pair in enumerate(zip(old_notice, new_notice)) if pair[0] != pair[1]] == [3], 'only notice source delta is oa-cli allowed-root check; default/body unchanged')
for p in ('scripts/package-linux.ps1', 'deployment/Dockerfile', 'deployment/Dockerfile.dockerignore'):
check(git(p).decode().splitlines() == (ROOT / p).read_text().splitlines(), 'Linux entry/deployment unchanged: ' + p)
old_windows = git('scripts/package-windows.ps1').decode().replace('\r\n', '\n')
common = (ROOT / 'scripts/package-common.ps1').read_text()
reader_old = old_windows.split('function Read-PeImports', 1)[1].split('\n# Windows system components', 1)[0].strip()
reader_new = common.split('function Read-PackagePeImports', 1)[1].split('\nfunction Get-PackageWindowsInputs', 1)[0].strip()
check(reader_old == reader_new, 'PE reader extraction is byte-equivalent after function rename/EOL normalization')
old_allowlist = re.search(r'\$systemDlls = @\((.*?)\n\)', old_windows, re.S)[1]
new_allowlist = re.search(r'\$systemDlls = @\((.*?)\n \)', common, re.S)[1]
check(re.findall(r"'([^']+)'", old_allowlist) == re.findall(r"'([^']+)'", new_allowlist), 'Windows system DLL allowlist unchanged')
new_windows = (ROOT / 'scripts/package-windows.ps1').read_text()
check("Get-PackageWindowsInputs $BinaryDirectory $Target $LlvmMinGWRoot @('arcadia-studio.exe', 'arcadia-server.exe')" in new_windows, 'Studio explicitly retains both original binary inputs')
check(old_windows.split('$evidence = Read-PackageBuildEvidence', 1)[1].splitlines() == new_windows.split('$evidence = Read-PackageBuildEvidence', 1)[1].splitlines(), 'Studio post-inspection packaging pipeline unchanged')
delta = subprocess.run(['git', 'diff', '--name-only', BASE, ASSEMBLY], cwd=ROOT, capture_output=True, check=True).stdout.decode().splitlines()
check(len(delta) == 9 and all(p.startswith('scripts/') for p in delta), 'assembly delta exactly nine packaging/script paths')
check(all((ROOT / p).read_bytes() == (SOURCE / p).read_bytes() for p in delta), 'reviewed main packaging sources equal frozen assembly bytes')
RESULT['source_review'] = {'semantic_source': BASE, 'assembly_source': ASSEMBLY, 'packaging_paths': {p: sha((ROOT / p).read_bytes()) for p in delta}, 'studio_document_counts': counts}
def artifact():
check(not PACKAGE.parent.exists(), 'fresh extraction directory')
check(ZIP.stat().st_size == 5091267 and sha(ZIP.read_bytes()) == ZIP_SHA, 'frozen ZIP size and hash')
with zipfile.ZipFile(ZIP) as z:
names = z.namelist()
check(len(names) == 114 and len({n.casefold() for n in names}) == 114, '114 unique ZIP files')
check(all(n.startswith(NAME + '/') and PurePosixPath(n).as_posix() == n and '..' not in PurePosixPath(n).parts and '\\' not in n and ':' not in n and all(not p.endswith((' ', '.')) for p in PurePosixPath(n).parts) for n in names), 'single safe normalized Windows ZIP root')
check(all(not i.is_dir() and stat.S_IFMT(i.external_attr >> 16) in (0, stat.S_IFREG) and i.date_time == (2000, 1, 1, 0, 0, 0) for i in z.infolist()) and z.testzip() is None, 'ordinary fixed-time ZIP members and CRCs')
PACKAGE.parent.mkdir()
z.extractall(PACKAGE.parent)
archived = {n[len(NAME) + 1:]: z.read(n) for n in names}
actual = snapshot(PACKAGE)
check(actual == archived, 'complete clean extraction equals ZIP bytes')
hashes = {p: sha(b) for p, b in actual.items()}
lines = actual['SHA256SUMS.txt'].decode().splitlines()
sums = dict((line[66:], line[:64]) for line in lines)
check(len(sums) == len(lines) and sums == {p: h for p, h in hashes.items() if p != 'SHA256SUMS.txt'}, 'complete checksum inventory equals every other extracted file')
package = read_json(PACKAGE / 'package.json')
generated = {'SHA256SUMS.txt', 'START-HERE.txt', 'THIRD-PARTY-NOTICES.txt', 'dependency-notices.json', 'documentation-check.json', 'package.json'}
check(set(actual) - set(package['input_sha256']) == generated and all(hashes[p] == h for p, h in package['input_sha256'].items()), 'all package inputs plus exactly six generated records')
check(hashes['arcadia.exe'] == EXE_SHA and hashes['libunwind.dll'] == DLL_SHA, 'frozen CLI and runtime hashes')
check(not {'arcadia-studio.exe', 'arcadia-server.exe'} & set(actual), 'no unrelated executable payload')
for p in actual:
if p.startswith(('docs/', 'examples/')) or p == 'LICENSE':
check((SOURCE / p).read_bytes() == actual[p], 'exact public document/example: ' + p)
check(actual['README.md'] == (SOURCE / 'scripts/package-cli-readme.md').read_bytes(), 'CLI-specific README')
notice = read_json(PACKAGE / 'dependency-notices.json')
check(notice['root_packages'] == ['oa-cli'] and len(notice['packages']) == 27, '27 dependency packages; only CLI root')
check(notice['cargo_lock_sha256'] == sha((SOURCE / 'Cargo.lock').read_bytes()) == '2b960242c33eb839ff008d01731f6d5cdb4e047ca8cd6e13151cbf9c09913510', 'original CLI Cargo lock identity')
lock = {(p['name'], p['version'], p.get('source')): p for p in tomllib.loads((SOURCE / 'Cargo.lock').read_text())['package']}
references = []
for p in notice['packages']:
check(lock[p['name'], p['version'], p['source']].get('checksum') == p['registry_checksum'], 'locked dependency: ' + p['name'])
references.extend(p['notices'])
for f in p['embedded_fonts']:
references.extend(f['notices'])
if p['source'] is None:
local = snapshot(SOURCE / 'crates' / p['name'])
local = {k: sha(v) for k, v in local.items() if 'target' not in PurePosixPath(k).parts and PurePosixPath(k).name != '.cargo-ok'}
check(local == p['local_source_sha256'], 'complete local crate inputs: ' + p['name'])
references += notice['rust']['notices'] + notice['runtime']['notices']
check(all(hashes[i['path']] == i['sha256'] for i in references), 'all package/compiler/runtime notice references and bytes')
check(notice['runtime']['sha256'] == DLL_SHA, 'runtime notice binds packaged DLL')
check(all(sha((SOURCE / p).read_bytes()) == h for p, h in notice['workspace_build_inputs_sha256'].items()), 'all shared build input hashes')
original = read_json(ROOT / '.local/source-cli-20260911/verified/manifest.json')
cli_inputs = [i for i in original['Artifacts'] if not i['Path'].startswith('.local/')]
check(len(cli_inputs) == 6 and all(sha((SOURCE / i['Path']).read_bytes()) == i['SHA256'].lower() for i in cli_inputs), 'six original CLI manifest inputs')
core_inputs = [i for i in read_json(ROOT / '.local/behavior-creation-20260911/gui-verified/manifest.json')['inputs'] if i['path'].startswith(('crates/oa-core/', 'build/'))]
check(len(core_inputs) == 95 and all(sha((SOURCE / i['path']).read_bytes()) == i['sha256'].lower() for i in core_inputs), '95 original core/shared build inputs')
RESULT['artifact'] = {'zip_sha256': ZIP_SHA, 'zip_bytes': ZIP.stat().st_size, 'file_count': len(actual), 'files': hashes, 'notice_packages': 27, 'notice_reference_files': len({i['path'] for i in references}), 'package_claims': {k: package[k] for k in ('product', 'version', 'target', 'acceptance', 'build', 'build_evidence')}}
return actual
def runtime(original_package):
run = OUT / 'runtime'
run.mkdir()
temp = run / 'temp'
temp.mkdir()
windows = Path(os.environ['SystemRoot'])
env = {'SystemRoot': str(windows), 'WINDIR': str(windows), 'PATH': ';'.join(map(str, (PACKAGE, windows / 'System32', windows))), 'TEMP': str(temp), 'TMP': str(temp)}
RESULT['environment'] = env
RESULT['invocations'] = []
def invoke(args, expect=0, report=True):
index = len(RESULT['invocations']) + 1
r = subprocess.run([str(PACKAGE / 'arcadia.exe'), *map(str, args)], cwd=run, env=env, capture_output=True, timeout=30, creationflags=subprocess.CREATE_NO_WINDOW)
(run / f'{index:02d}.stdout').write_bytes(r.stdout)
(run / f'{index:02d}.stderr').write_bytes(r.stderr)
RESULT['invocations'].append({'arguments': list(map(str, args)), 'exit': r.returncode, 'stdout_sha256': sha(r.stdout), 'stderr_sha256': sha(r.stderr)})
check(r.returncode == expect, 'CLI expected exit: ' + str(args))
if not report:
return r.stdout
check(not r.stderr, 'JSON invocation keeps diagnostic envelope on stdout')
value = json.loads(r.stdout)
check(value['format_version'] == 1 and value['exit_code'] == expect, 'versioned JSON report agrees with process exit')
return value
check(invoke(['--version'], report=False).decode().strip() == 'arcadia 0.2.0', 'underlying CLI version')
help_text = invoke(['--help'], report=False).decode()
check(all(word in help_text for word in ('check ROOT', 'stabilize ROOT', 'rename-alias ROOT OLD NEW', '--write', 'provisional')), 'packaged help has command/preview contract')
examples = run / 'examples'
shutil.copytree(PACKAGE / 'examples', examples)
before_examples = snapshot(examples)
for relative in ('text-first/project.arcadia', 'release1/project.arcadia'):
r = invoke(['check', examples / relative, '--json'])
check(r['status'] == 'checked' and not r['written'] and not r['changes'], 'packaged example check: ' + relative)
check(snapshot(examples) == before_examples, 'checks preserve complete copied examples byte-for-byte')
fixture = run / 'exact-source'
(fixture / 'modules').mkdir(parents=True)
entry = 'arcadia 7;\r\nproject review "CLI package 救助" id "c073d2cf-83e5-4c98-8eaf-4a465ff0cf71" {\r\n description "Literal @crew remains";\r\n include "modules/oa.arcadia";\r\n}\r\n'
module = '// @crew stays in this comment; Ω\r\noa_actor crew "Crew literal @crew" id "1987c990-e3cb-4f0e-9be9-fe61f007b997" {}\r\noa_activity observe "Observe Ω" {}\r\nallocation responsibility "Assigned @crew" id "ffcf7e41-f94b-4f7c-99c1-1a30a0b46959" { source @crew; target @observe; }\r\n'
(fixture / 'project.arcadia').write_bytes(entry.encode())
(fixture / 'modules/oa.arcadia').write_bytes(module.encode())
(fixture / 'detached.arcadia').write_bytes(b'// known detached bytes remain\r\n')
before = snapshot(fixture)
root = fixture / 'project.arcadia'
def stabilized(report):
check(len(report['changes']) == 1 and report['changes'][0]['path'] == 'modules/oa.arcadia' and report['changes'][0]['before'] == module, 'only missing-ID source changes; exact captured raw bytes')
text = report['changes'][0]['after']
match = re.search(r'oa_activity observe "Observe Ω" id "([0-9a-f-]{36})"', text)
check(match is not None, 'fresh ID is attached to literal intended activity')
identity = str(uuid.UUID(match[1]))
check(identity not in (entry + module), 'allocated identity differs from every preexisting identity')
expected = module.replace('oa_activity observe "Observe Ω" {}', f'oa_activity observe "Observe Ω" id "{identity}" {{}}')
check(text == expected, 'stabilization preserves every other character, CRLF, comment, name and reference')
return text, identity
preview = invoke(['stabilize', root, '--json'])
check(preview['status'] == 'preview' and not preview['written'] and preview['identities_to_stabilize'] == 1, 'stabilization is preview by default')
stabilized(preview)
check(snapshot(fixture) == before, 'preview preserves all physical source bytes including detached file')
written = invoke(['stabilize', root, '--write', '--json'])
check(written['status'] == 'written' and written['written'], 'explicit stabilization writes')
text, identity = stabilized(written)
stable_expected = dict(before, **{'modules/oa.arcadia': text.encode()})
check(snapshot(fixture) == stable_expected, 'stabilized complete physical file inventory and bytes')
renamed = text.replace('oa_actor crew ', 'oa_actor rescue_team ').replace('source @crew;', 'source @rescue_team;')
preview = invoke(['rename-alias', root, 'crew', 'rescue_team', '--json'])
check(not preview['written'] and preview['status'] == 'preview' and [(c['path'], c['before'], c['after']) for c in preview['changes']] == [('modules/oa.arcadia', text, renamed)], 'alias preview changes exact declaration and typed reference only')
check(snapshot(fixture) == stable_expected, 'rename preview retains exact source bytes and identities')
saved = invoke(['rename-alias', root, 'crew', 'rescue_team', '--write', '--json'])
renamed_expected = dict(before, **{'modules/oa.arcadia': renamed.encode()})
check(saved['written'] and snapshot(fixture) == renamed_expected, 'explicit alias write matches full independent raw expected files')
expected_ids = re.findall(r'\bid "([0-9a-f-]{36})"', entry + text)
check(re.findall(r'\bid "([0-9a-f-]{36})"', entry + renamed) == expected_ids and len(expected_ids) == 4, 'all four identities remain ordered and exact through rename')
timestamps = {p: (fixture / p).stat().st_mtime_ns for p in renamed_expected}
noop = invoke(['stabilize', root, '--write', '--json'])
check(noop['status'] == 'unchanged' and not noop['written'] and not noop['changes'] and noop['identities_to_stabilize'] == 0, 'explicit no-op reports unchanged')
check(snapshot(fixture) == renamed_expected and timestamps == {p: (fixture / p).stat().st_mtime_ns for p in renamed_expected}, 'no-op preserves all file bytes and mtimes')
invalid = run / 'invalid.arcadia'
invalid.write_bytes(b'arcadia 7;\r\nproject broken "Unclosed" {\r\n')
invalid_before = invalid.read_bytes()
refusal = invoke(['stabilize', invalid, '--write', '--json'], expect=1)
check(not refusal['written'] and invalid.read_bytes() == invalid_before, 'invalid source write refuses with exact raw retention')
check(snapshot(PACKAGE) == original_package and sha(ZIP.read_bytes()) == ZIP_SHA, 'all packaged files and frozen ZIP unchanged after runtime checks')
RESULT['runtime'] = {'fixed_source_ids': expected_ids, 'allocated_activity_id': identity, 'final_source_sha256': {p: sha(b) for p, b in renamed_expected.items()}, 'source_exact_expected': {p: b.decode() for p, b in renamed_expected.items()}, 'stabilization_note': 'Separate preview UUID remains provisional; each result is checked by literal declaration role, not assumed equal across invocations.'}
try:
check(not (OUT / 'result.json').exists(), 'preserve earlier evidence; no overwrite')
source_checks()
frozen = artifact()
runtime(frozen)
RESULT['status'] = 'PASS'
except Exception as error:
RESULT['status'] = 'FAIL'
RESULT['error'] = repr(error)
RESULT['traceback'] = traceback.format_exc()
RESULT['finished_utc'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
RESULT['script_sha256'] = sha(Path(__file__).read_bytes())
record(OUT / 'result.json', RESULT)
print(json.dumps({k: RESULT[k] for k in ('status', 'started_utc', 'finished_utc')}, indent=2))
print('checks:', len(RESULT['checks']), 'invocations:', len(RESULT.get('invocations', [])))
if RESULT['status'] == 'FAIL':
print(RESULT['traceback'])
sys.exit(1)