156 lines
8.1 KiB
Python
156 lines
8.1 KiB
Python
#!/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())
|