"""Data-only public-read reconciliation; run only after parent freezes all gates. Usage: python -I -B inspect_runtime.py REVIEWED_FINAL_SOURCE_SHA Runs only read-only Git commands, never Cargo, archived helpers or executables. """ from pathlib import Path import ast import hashlib import io import json import re import subprocess import sys import zipfile assert not sys.flags.optimize assert len(sys.argv) == 2 source = sys.argv[1] assert re.fullmatch(r'[0-9a-f]{40}', source) base = Path(__file__).resolve().parents[2] root = base / '.local/as3batch' folder = base / '.local/analysis-public-read-gates-20260925' destination = Path(__file__).with_name('PUBLIC-READ-RUNTIME-READBACK.json') assert not destination.exists(), 'Preserve prior readback; no overwrite' sha = lambda b: hashlib.sha256(b).hexdigest() GATE_TEMPLATE = '9088952ca3fabf56ab4434273bf55c6ec6eb6139bc2452a7be7c66a4409e420d' CLI_HASH = '25ec9ab5aa22d7364169772eae2cc10d508415fada9845a5fb7f115e7dd3f3e0' ARCHIVE_HASH = '879ec5a2198898a8a6d792d163710ca1d6ca83f8e7773ee093dcc93456c77378' BASELINE = '7b06b8ea1b7ef902ee0b3d0ec2971a9a7a1f1acd' DOC_SOURCE = 'bab6f96a5b05b37f2b5d67a430aed0f67d783e73' DOC_LOG = 'c93d7bfc96884dcbfb54aa2a36ba2294fd48bab036fb676931c54b7b8fd806d2' DOC_PREFIX = (r'crates\oa-core\src\language\analysis_source\mixed\project.rs - ' 'language::analysis_source::mixed::project::') DOC_POSITIVE = DOC_PREFIX + 'check_mixed_bundle (line 87)' DOC_MAPPING = { DOC_PREFIX + 'CheckedMixed (line 36) - compile fail': DOC_PREFIX + 'CheckedMixed (line 36)', DOC_PREFIX + 'CheckedMixed (line 50) - compile fail': DOC_PREFIX + 'CheckedMixed (line 50)', } DOC_RECORDER_FAILURE = ['Doc runtime names must exactly match the prior nonzero listing'] BRIDGE = ('language::analysis_source::mixed::carried::public_read::' 'public_read_fresh_constructor_keeps_zero_carry_both_modes_and_exact_policy') EXTERNAL = { 'public_mixed_read_complete_oa_sa_models_and_all_located_warnings', 'public_mixed_read_scoped_uuid_reuse_and_cross_file_call_are_not_global_collisions', 'public_mixed_read_explicit_identity_classes_versions_and_long_forms', 'public_mixed_read_exact_input_file_edge_and_depth_boundaries', 'public_mixed_read_paths_missing_cycles_and_detached_files_are_errors', 'public_mixed_read_final_nested_model_limit_is_distinct_from_syntax_entries', 'public_mixed_read_typed_reference_caller_allowance_has_a_real_positive_control', 'public_mixed_read_malformed_duplicate_and_wrong_domain_references', 'public_mixed_read_unicode_columns_are_not_utf8_byte_columns', 'public_mixed_read_ordinary_parity_uses_an_independent_complete_model', } SCOPE = ['Cargo.toml', 'Cargo.lock', 'rust-toolchain.toml', '.cargo', 'build', 'crates', 'assets', 'docs', 'examples', 'harness/fixtures', 'vendor'] def git(*args, data=None): return subprocess.check_output(['git', '--no-replace-objects', *args], cwd=root, input=data) def within(parent, name): p = (parent / name).resolve() assert p.is_relative_to(parent.resolve()), name return p def rows(raw): result = re.findall(r'^test ([^\r\n]+) \.\.\. (ok|FAILED|ignored)\r?$', raw.decode('utf8'), re.M) assert len(result) == len(dict(result)), 'Duplicate runtime name' return result def names(raw): result = re.findall(r'^([^\r\n]+): test\r?$', raw.decode('utf8'), re.M) assert len(result) == len(set(result)), 'Duplicate listed name' return result def all_passed(raw, count, grouped=False): actual = rows(raw) assert len(actual) == count and all(status == 'ok' for _, status in actual) summaries = re.findall(r'^test result: ok\. (\d+) passed; 0 failed; 0 ignored; 0 measured; (\d+) filtered out;', raw.decode('utf8'), re.M) assert summaries and sum(int(x[0]) for x in summaries) == count if not grouped: assert len(summaries) == 1 return {name for name, _ in actual} archive_raw = (base / 'harness/experiments/ordinary-source8-20260925/evidence.zip').read_bytes() assert sha(archive_raw) == ARCHIVE_HASH archive = zipfile.ZipFile(io.BytesIO(archive_raw)) assert archive.testzip() is None def prior(stem): r = archive.read('gates/' + stem + '.result.json') d = json.loads(r) log = archive.read('gates/' + stem + '.log') assert sha(log) == d['log_sha256'] assert rows(log) == [tuple(x) for x in d['outcomes']] return d, log old, old_log = prior('ordinary-full-0c07f53-1') old_names = all_passed(old_log, 298) old_listing, old_list_log = prior('ordinary-list-0c07f53-1') assert set(names(old_list_log)) == old_names old_fmt, old_fmt_log = prior('ordinary-fmt-7b06b8e-1') assert old_fmt['exit'] == 1 and sha(old_fmt_log) == 'c1c6b12861c8117ce6ffc6d6742ebf92618aafb10160ec6f99d370ca72ebab2d' environment = dict(old['environment']) environment.update(RUSTDOC=None, RUSTDOCFLAGS=None, CARGO_ENCODED_RUSTDOCFLAGS=None) assert environment['CARGO_BUILD_JOBS'] == '1' and environment['CARGO_INCREMENTAL'] == '0' for key in ['RUSTFLAGS', 'RUSTC', 'RUSTUP_TOOLCHAIN', 'ARCADIA_CODEC_LIFECYCLE_PHASE', 'RUST_MIN_STACK', 'RUSTDOC', 'RUSTDOCFLAGS', 'CARGO_ENCODED_RUSTDOCFLAGS']: assert environment[key] is None assert git('rev-parse', 'HEAD').decode().strip() == source assert not git('status', '--porcelain', '--untracked-files=all').strip() input_cache = {} def inputs_at(commit): if commit in input_cache: return input_cache[commit] paths = git('ls-tree', '-r', '--name-only', commit, *SCOPE).decode().splitlines() data = git('cat-file', '--batch', data=''.join(f'{commit}:{p}\n' for p in paths).encode()) stream = io.BytesIO(data) result = [] for path in paths: header = stream.readline().split() assert header[1] == b'blob' raw = stream.read(int(header[2])) assert stream.read(1) == b'\n' if commit == source: assert within(root, path).read_bytes() == raw, path result.append({'path': path, 'bytes': len(raw), 'sha256': sha(raw)}) assert stream.read() == b'' input_cache[commit] = result return result expected_inputs = inputs_at(source) verified = {} def read_gate(stem, *, current=True): receipt_raw = (folder / (stem + '.result.json')).read_bytes() d = json.loads(receipt_raw) raw = (folder / (stem + '.log')).read_bytes() assert sha(raw) == d['log_sha256'] assert d['source'] == d['head_after'] assert d['cwd'] == str(root) and d['invocation'] == 'cargo' assert d['environment'] == environment assert not d['changed_inputs'] and d['source_status_after'] == '' manifest = within(folder, d['inputs']).read_bytes() assert sha(manifest) == d['inputs_sha256'] assert json.loads(manifest) == {'source': d['source'], 'inputs': inputs_at(d['source'])} launcher = within(folder, d['launcher_copy']).read_bytes() assert sha(launcher) == d['launcher_sha256'] == d['launcher_sha256_after'] assert launcher.count(d['source'].encode()) == 1 assert sha(launcher.replace(d['source'].encode(), b'PENDING_REVIEWED_PUBLIC_READ_SOURCE')) == GATE_TEMPLATE assert rows(raw) == [tuple(x) for x in d['outcomes']] listing = names(raw) if d['mode'] in ('list', 'integration-list', 'doc-list') else [] assert listing == d['listing_members'] and len(listing) == d['listing_count'] if current: assert d['source'] == source if stem == 'public-read-doc-suite-bab6f96-1': # Exact observed rustdoc spelling only; retain the recorder's refusal. assert source == DOC_SOURCE and sha(raw) == DOC_LOG and d['exit'] == 0 assert d['verification_failures'] == DOC_RECORDER_FAILURE else: assert not d['verification_failures'] for artifact in d['artifacts']: path = Path(artifact['path']) assert path.resolve().is_relative_to(Path(environment['CARGO_TARGET_DIR']).resolve()) binary = path.read_bytes() assert len(binary) == artifact['bytes'] and sha(binary) == artifact['sha256'] verified[stem] = {'source': d['source'], 'exit': d['exit'], 'receipt_sha256': sha(receipt_raw), 'log_sha256': sha(raw), 'inputs_sha256': sha(manifest), 'launcher_sha256': sha(launcher), 'outcomes': d['outcomes'], 'listing': listing, 'artifacts': d['artifacts'], 'verification_failures': d['verification_failures']} return d, raw def current_gate(mode, command, target=None, selected_filter='', artifact_count=1): stem = 'public-read-' + mode + ('-' + target if target else '') + '-' + source[:7] + '-1' d, raw = read_gate(stem) assert d['mode'] == mode and d['command'] == command assert d['test_target'] == target and d['filter'] == selected_filter assert d['exit'] == (1 if mode == 'fmt' else 0) assert len(d['artifacts']) == artifact_count if artifact_count and mode != 'cli-build': running = re.findall(r'Running [^\r\n]+ \(([^\r\n]+\.exe)\)', raw.decode()) assert len(running) == artifact_count assert [(root / n).resolve() for n in running] == [Path(a['path']).resolve() for a in d['artifacts']] return d, raw prefix = ['cargo', 'test', '--locked', '--offline', '-p', 'oa-core', '--lib'] listing, listing_log = current_gate('list', prefix + ['--', '--list']) focused, focused_log = current_gate('focused', prefix + ['public_read', '--', '--test-threads=1'], selected_filter='public_read') full, full_log = current_gate('full', prefix + ['--', '--test-threads=1']) assert all_passed(full_log, 299) == old_names | {BRIDGE} assert set(listing['listing_members']) == old_names | {BRIDGE} assert all_passed(focused_log, 1) == {BRIDGE} assert listing['artifacts'] == focused['artifacts'] == full['artifacts'] clippy, _ = current_gate('clippy', ['cargo', 'clippy', '--locked', '--offline', '-p', 'oa-core', '--all-targets', '--', '-D', 'warnings'], artifact_count=0) fmt, fmt_log = current_gate('fmt', ['cargo', 'fmt', '--all', '--', '--check'], artifact_count=0) assert fmt_log == old_fmt_log suites = {'mixed_source8_read': '', 'source_project': '', 'source_identity_updates': '', 'source_nested_updates': '', 'source_files': '', 'r3_migration': '', 'review_regressions': 'checked_save_', 'source7_contract': 'source7_source_bundle_migration_is_token_only_and_new_link_stays_with_caller'} integration = {} for target, selected_filter in suites.items(): pair = {} for mode in ('integration-list', 'integration-suite'): command = ['cargo', 'test', '--locked', '--offline', '-p', 'oa-core', '--test', target] if selected_filter: command.append(selected_filter) command += ['--', '--list' if mode == 'integration-list' else '--test-threads=1'] if target == 'source7_contract': command.append('--exact') pair[mode] = current_gate(mode, command, target, selected_filter) l, s = pair['integration-list'][0], pair['integration-suite'][0] assert l['artifacts'] == s['artifacts'] assert Path(s['artifacts'][0]['path']).name.startswith(target + '-') if target == 'mixed_source8_read': expected = EXTERNAL else: checkpoint = '7b06b8e' if target == 'r3_migration' else '0c07f53' previous, previous_log = prior(f'ordinary-integration-suite-{target}-{checkpoint}-1') expected = all_passed(previous_log, len(previous['outcomes'])) assert set(l['listing_members']) == expected assert all_passed(pair['integration-suite'][1], len(expected)) == expected integration[target] = sorted(expected) assert sum(len(v) for k, v in integration.items() if k != 'mixed_source8_read') == 65 assert 'deep_structural_writeback_runs_on_a_small_thread_stack' in integration['source_nested_updates'] doc_prefix = ['cargo', 'test', '--locked', '--offline', '-p', 'oa-core', '--doc'] doc_list, _ = current_gate('doc-list', doc_prefix + ['--', '--list'], artifact_count=0) doc_suite, doc_log = current_gate('doc-suite', doc_prefix + ['--', '--test-threads=1'], artifact_count=0) doc_names = set(doc_list['listing_members']) assert source == DOC_SOURCE assert doc_names == {DOC_POSITIVE, *DOC_MAPPING.values()} doc_runtime_names = all_passed(doc_log, 3, grouped=True) assert doc_runtime_names == {DOC_POSITIVE, *DOC_MAPPING} assert {DOC_MAPPING.get(n, n) for n in doc_runtime_names} == doc_names assert re.findall(r'^test result: ok\. (\d+) passed; 0 failed; 0 ignored; 0 measured; (\d+) filtered out;', doc_log.decode('utf8'), re.M) == [('1', '0'), ('2', '0')] assert doc_suite['verification_failures'] == DOC_RECORDER_FAILURE public_docs = {n for n in doc_names if 'CheckedMixed' in n or 'check_mixed_bundle' in n} assert len(public_docs) == 3, public_docs build, build_log = current_gate('cli-build', ['cargo', 'build', '--locked', '--offline', '-p', 'oa-cli', '--bin', 'arcadia', '--message-format=json']) emitted = [] for line in build_log.splitlines(): try: row = json.loads(line) except (json.JSONDecodeError, UnicodeDecodeError): continue if row.get('reason') == 'compiler-artifact' and row.get('target', {}).get('name') == 'arcadia' and row.get('executable'): assert row['target']['kind'] == ['bin'] and not row['profile']['test'] emitted.append(row['executable']) assert len(emitted) == 1 and Path(emitted[0]).resolve() == Path(build['artifacts'][0]['path']).resolve() original = {name: archive.read('cli/original-inputs/' + name).decode('utf8') for name in ['project.arcadia', 'participants.arcadia', 'children.arcadia']} after = {'participants.arcadia': original['participants.arcadia'].replace('oa_entity team ', 'oa_entity rescue_team '), 'children.arcadia': original['children.arcadia'].replace('parent @team;', 'parent @rescue_team;')} inventory = {name: {'bytes': len(text.encode('utf8')), 'sha256': sha(text.encode('utf8'))} for name, text in original.items()} fixture = folder / ('cli-fixture-' + source[:7]) assert sorted(p.relative_to(fixture).as_posix() for p in fixture.rglob('*')) == sorted(original) for name, text in original.items(): assert (fixture / name).read_bytes() == text.encode('utf8') for operation in ('check', 'rename-alias'): stem = f'public-read-cli-{operation}-{source[:7]}-1' receipt_raw = (folder / (stem + '.cli.json')).read_bytes() d = json.loads(receipt_raw) stdout = (folder / (stem + '.stdout.json')).read_bytes() stderr = (folder / (stem + '.stderr.txt')).read_bytes() assert d['source'] == source and d['build_artifact'] == build['artifacts'][0] assert d['cwd'] == str(root) and d['environment'] == environment assert d['inputs'] == build['inputs'] and d['inputs_sha256'] == build['inputs_sha256'] assert not d['changed_inputs'] and not d['verification_failures'] and d['exit'] == 0 assert d['stdout_sha256'] == sha(stdout) and d['stderr_sha256'] == sha(stderr) and stderr == b'' assert d['expected_input_inventory'] == d['after_inventory'] == inventory launcher = within(folder, d['launcher_copy']).read_bytes() assert sha(launcher) == d['launcher_sha256'] == CLI_HASH entry = str(fixture / 'project.arcadia') argv = [build['artifacts'][0]['path'], operation, entry] if operation == 'rename-alias': argv += ['team', 'rescue_team'] assert d['argv'] == argv + ['--json'] report = json.loads(stdout) for key, value in {'format_version': 1, 'entry': entry, 'operation': operation, 'status': 'checked' if operation == 'check' else 'preview', 'exit_code': 0, 'written': False, 'diagnostics': [], 'identities_to_stabilize': 0, 'migration': None}.items(): assert report[key] == value if operation == 'check': assert report['changes'] == [] else: assert [v['path'] for v in report['changes']] == ['children.arcadia', 'participants.arcadia'] for v in report['changes']: assert v['before'] == original[v['path']] and v['after'] == after[v['path']] verified[stem] = {'receipt_sha256': sha(receipt_raw), 'stdout_sha256': sha(stdout), 'stderr_sha256': sha(stderr), 'launcher_sha256': sha(launcher)} # Preserve all actual earlier gate attempts without calling overwritten artifacts current. historical = {} for path in sorted(folder.glob('*.result.json')): if path.stem.removesuffix('.result') in verified: continue stem = path.name.removesuffix('.result.json') d, raw = read_gate(stem, current=False) historical[stem] = verified[stem] initial = historical['public-read-integration-list-mixed_source8_read-ddf823a-1'] assert initial['exit'] == 101 and not initial['outcomes'] and not initial['artifacts'] and not initial['listing'] assert initial['log_sha256'] == '3acdd5b0cf1dbd40c08c7e1c6feacd582ac152f3e911018223551d4e718aba51' assert initial['verification_failures'] == ['Expected nonzero unique target listing', 'Expected exactly one requested integration target executable'] # Exact stack names were independently enumerated in RUNTIME-PLAN; no inference of headroom. plan = Path(__file__).with_name('RUNTIME-PLAN.md').read_text(encoding='utf8') stack_names = re.findall(r'^- `(language::[^`]+)`$', plan, re.M) assert len(stack_names) == len(set(stack_names)) == 11 assert set(stack_names) <= old_names and all(dict(full['outcomes'])[n] == 'ok' for n in stack_names) assert git('rev-parse', 'HEAD').decode().strip() == source assert not git('status', '--porcelain', '--untracked-files=all').strip() result = {'source': source, 'baseline': BASELINE, 'baseline_archive_sha256': ARCHIVE_HASH, 'checker_sha256': sha(Path(__file__).read_bytes()), 'plan_sha256': sha(Path(__file__).with_name('RUNTIME-PLAN.md').read_bytes()), 'inputs': len(expected_inputs), 'prior_passes_preserved': 298, 'new_bridge': BRIDGE, 'full_passed': 299, 'external_names': sorted(EXTERNAL), 'integration': integration, 'doc_names': sorted(doc_names), 'doc_runtime_names': sorted(doc_runtime_names), 'doc_exact_mapping': DOC_MAPPING, 'doc_recorder_verification_failures': DOC_RECORDER_FAILURE, 'public_docs': sorted(public_docs), 'stack_controls': stack_names, 'core_artifact': full['artifacts'][0], 'cli_artifact': build['artifacts'][0], 'environment': environment, 'inherited_fmt_sha256': sha(fmt_log), 'verified': verified, 'historical': historical, 'scope': 'Data/Git readback only; no compiler/helper/product execution. Local-only Windows evidence; no Linux/depth128 or public mixed CLI qualification.'} with destination.open('x', encoding='utf8', newline='\n') as stream: json.dump(result, stream, indent=2) stream.write('\n') print(json.dumps({'source': source, 'full_passed': 299, 'external_passed': 10, 'ordinary_integration_passed': 65, 'doctests': len(doc_names), 'readback_sha256': sha(destination.read_bytes())}))