#!/usr/bin/env python3 """Read-only P22 record integrity; exit 2 for unmet original readiness.""" import argparse import hashlib import json from pathlib import Path import struct import sys ROOT = Path(__file__).resolve().parents[1] OUT = 'docs/plans/original-completion-2026-10-02/execution/p22-02/' INPUT = 'docs/spec/inputs/P22-02-accepted-usage.json' CURRENT = 'docs/plans/original-completion-2026-10-02/checkpoints/p22-02-slot-contract-r02/all-slot-current-candidate-contracts.json' SPEC_SHA = 'f252de3c5cff7ebfa62322ae2f2df09fb80a44bc59eb286b76d4acb4c12ca875' INPUT_SHA = '45fa52acf0c8bd4dd58c3b40b681cb92066f05d617df157fc18370a5e8816281' def need(ok, message): if not ok: raise ValueError(message) def sha(raw): return hashlib.sha256(raw).hexdigest() def semantic(value): return sha(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode()) def pairs(items): result = {} for key, value in items: need(key not in result, 'duplicate JSON key: ' + key) result[key] = value return result def read(path): return json.loads((ROOT / path).read_text(), object_pairs_hook=pairs) def decode(doc, node=None): need(type(doc) is dict and all(type(doc.get(k)) is list for k in ('keys', 'strings', 'pool')), 'wrong codec tables') need(all(type(k) is str for k in doc['keys']) and len(set(doc['keys'])) == len(doc['keys']), 'duplicate/wrong key table') active = set() def index(values, i): need(type(i) is int and 0 <= i < len(values), 'bad reference index') return values[i] def walk(n): if not isinstance(n, list): need(n is None or type(n) in (str, int, float, bool), 'wrong scalar type') return n need(bool(n) and type(n[0]) is str, 'wrong node type') tag = n[0] if tag == 'l': return [walk(x) for x in n[1:]] if tag == 'd': need(len(n) % 2 == 1, 'malformed flat dictionary') result = {} for i in range(1, len(n), 2): key = index(doc['keys'], n[i]) need(type(key) is str and key not in result, 'duplicate or invalid decoded key') result[key] = walk(n[i + 1]) return result need(tag in ('s', 'r') and len(n) == 2, 'unknown node') if tag == 's': value = index(doc['strings'], n[1]) need(type(value) is str, 'wrong shared string type') return value i = n[1] value = index(doc['pool'], i) need(i not in active and isinstance(value, list) and value[0] in ('d', 'l'), 'cycle or wrong shared object type') active.add(i) result = walk(value) active.remove(i) return result result = walk(doc['files'] if node is None else node) for i in range(len(doc['pool'])): walk(['r', i]) return result def check(path): spec = (ROOT / 'docs/spec/acceptance/P22-02.json').read_bytes() need(sha(spec.decode().strip().encode()) == SPEC_SHA, 'contract spec SHA mismatch') need(sha((ROOT / INPUT).read_bytes()) == INPUT_SHA, 'accepted input raw SHA mismatch') accepted = read(INPUT) original = decode(accepted) records = {} for name, rows in original.items(): need(type(rows) is list, 'decoded file must be a list') pin = accepted['source_files'][name] need(len(rows) == pin['records'] and semantic(rows) == pin['semantic_sha256'], 'decoded semantic file mismatch: ' + name) for row in rows: identity = row.get('slot_id', row.get('group_id')) need(type(identity) is str and identity not in records, 'duplicate or invalid identity') records[identity] = (name, row) need(len(records) == 614, 'accepted identity coverage') packed = read(path) ledger = decode(packed) need(ledger['accepted_input'] == accepted, 'lossless accepted input mismatch') current = read(CURRENT) need(ledger['current'] == current, 'current contract fields changed/omitted or false promotion') evidence_dir = ROOT / 'docs/plans/original-completion-2026-10-02/checkpoints/p22-02-current-contract-r01' expected_evidence = {CURRENT} | {str(p.relative_to(ROOT)) for p in evidence_dir.iterdir() if p.is_file()} need(set(ledger['evidence_pins']) == expected_evidence, 'evidence omission/extra') selection = ledger['selection_contract'] need(selection['selected_assets'] == [], 'unsupported selected asset promotion') need('D08/D09 UNKNOWN' in selection['quantity'] and '2 state groups are not images' in selection['quantity'], 'invented quantities/state image promotion') condition_keys = {'all_exact_uses_selected_and_usable', 'all_creator_and_rights_evidence_accepted', 'all_naming_resolution_layout_replacement_references_accepted', 'all_mandatory_unmade_and_disclosure_gaps_resolved', 'applicable_display_and_hearing_evidence', 'independent_opus_review', 'root_static_and_meaningful_negative_validation', 'integration_and_final_original_30_88_acceptance'} need(set(ledger['original_conditions']) == condition_keys and all(v is False for v in ledger['original_conditions'].values()), 'unsupported original evidence promotion') for evidence, digest in ledger['evidence_pins'].items(): need(sha((ROOT / evidence).read_bytes()) == digest, 'evidence raw SHA mismatch: ' + evidence) seen = set() used = set() for contract in current['contracts']: identity = contract['id'] need(identity in records and identity not in seen, 'contract identity coverage/duplicate') seen.add(identity) name, record = records[identity] need(contract['source_file'] == name and semantic(record) == contract['accepted_record_semantic_sha256'], 'accepted record mismatch') for anchor in contract['source_anchors']: raw = (ROOT / anchor['path']).read_bytes() need(sha(raw) == anchor['sha256'], 'anchor raw SHA') need(raw.decode().splitlines()[anchor['line'] - 1] == anchor['text'], 'anchor line/reference mismatch') for ref in contract['candidate_refs']: source = ref['candidate_ref'] need(source in current['candidate_pool'], 'omitted candidate source') used.add(source) actual = current['candidate_pool'][source]['current_raw_sha256'] need(ref['prior_source_still_equal'] == (ref['prior_raw_sha256'] == actual), 'historical applicability mismatch') need(seen == set(records), 'omitted identities') need(used == set(current['candidate_pool']) and len(used) == 19, 'source coverage') for source, fact in current['candidate_pool'].items(): raw = (ROOT / source).read_bytes() need(sha(raw) == fact['current_raw_sha256'], 'current source raw SHA: ' + source) fmt = fact['observed_format'] if source.endswith('.png'): need(raw[:8] == b'\x89PNG\r\n\x1a\n', 'PNG signature') width, height = struct.unpack('>II', raw[16:24]) need((width, height) == (fmt['png_width'], fmt['png_height']), 'dimension mismatch') pending = [c['id'] for c in current['contracts'] if not c['original_asset_completion'] or c['missing_conditions'] or c['finished_asset'] is None] # Nonvacuous: all accepted uses plus external original acceptance conditions. ready = bool(seen) and not pending and all(ledger['original_conditions'].values()) return {'record_integrity': True, 'original_unit_ready': ready, 'pending_identity_count': len(pending), 'original_conditions': ledger['original_conditions'], 'acceptance_claimed': False} def main(): parser = argparse.ArgumentParser() parser.add_argument('--ledger', default=OUT + 'asset-contracts.json') parser.add_argument('--require-ready', action='store_true') args = parser.parse_args() try: result = check(args.ledger) print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 2 if args.require_ready and not result['original_unit_ready'] else 0 except (ValueError, KeyError, TypeError, IndexError, OSError, RecursionError) as exc: print(json.dumps({'record_integrity': False, 'error': str(exc), 'acceptance_claimed': False})) return 1 if __name__ == '__main__': sys.exit(main())