VPL Build FrameworkPortable architecture for building, testing, reviewing, releasing and operating AI-enabled products quickly without sacrificing traceability or control.

15,710 bytes · SHA-256 81a0620364bae8771314b4e8a21010735c9db39a86298f363934b3194305420e

Download exact source file

#!/usr/bin/env python3
"""Apply only the approved R03 documentary addendum; never alter frozen source.

Exit 0 means deterministic composition/validation succeeded, NOT semantic approval.
The unchanged R01 verifier is used for baseline source checks only, without any
adjudication. No candidate, Action 3/4, or old unit-test suite is executed.
"""
from __future__ import annotations
import argparse, copy, hashlib, importlib.util, json, sys, zipfile
from pathlib import Path

PRIMARY_SHA = 'd0054ef48173343dc3ba7d08b1a3bb031019b3f0b459c44f8546ae86a27d2c8b'
BASE_RECEIPT_SHA = '3893f7cae4a710cb2fd20de405e769b419e5d46b9775c6c55022cbab3a03c3fb'
GL = 'controls/VPL_V5_06_GUIDELINE_REGISTRY_R02.json'
AGENT = 'controls/VPL_V5_06_AGENT_REGISTRY_R02.json'
O1 = 'O1_ADOPTED_AMENDMENTS.md'
V4_GL = 'Treat model persistence as untrusted. Load every canonical V4 operating control with equal persistence priority; evaluate trigger/risk/applicability only after complete load. Prove semantic completeness with manifest/receipt, balanced challenge coverage and immediate behavioral test. Self-report, recap or selective retrieval is not evidence. Consequential mutation requires PERSISTENCE_STATE=CURRENT.'
APPROVAL = 'Approve replacing the A01/M26 mandatory-when-proven 15-minute Hostinger producer with invocation-bound V5 persistence. Cron is optional and cannot establish CURRENT. Retain every V4 GL-15 safeguard; V5 GL-15 is additive. This is design approval only, not deployment or score approval.'
CLAUSES = ['Treat model persistence as untrusted.', 'immediate behavioral test', 'selective retrieval is not evidence']

class Rejected(ValueError): pass

def require(value, message):
    if not value: raise Rejected(message)

def sha(b): return hashlib.sha256(b).hexdigest()
def canonical(x): return (json.dumps(x, sort_keys=True, ensure_ascii=False, allow_nan=False, separators=(',', ':'))+'\n').encode()
def pretty(x): return (json.dumps(x, indent=2, ensure_ascii=False, allow_nan=False)+'\n').encode()

def load_json(raw):
    def pairs(items):
        d={}
        for k,v in items:
            require(k not in d, 'duplicate JSON key: '+k); d[k]=v
        return d
    def bad(v): raise Rejected('nonfinite JSON number: '+v)
    return json.loads(raw, object_pairs_hook=pairs, parse_constant=bad)

def row(obj, array, id):
    a=[r for r in obj[array] if r.get('id')==id]
    require(len(a)==1, 'missing/duplicate row: '+id)
    return a[0]

def load_baseline(root: Path):
    """Verify reviewed archive and every expanded inner member before import."""
    raw=(root/'originals/VPL_ACTION1_SEMANTIC_CLAUDE_READY_R01.zip').read_bytes()
    require(sha(raw)==PRIMARY_SHA,'wrong reviewed primary archive')
    base=root/'prior_reviewed_R01'
    prefix='ACTION1_REVIEW_DELIVERY_R01/ACTION1_SEMANTIC_R01/'
    with zipfile.ZipFile(root/'originals/VPL_ACTION1_SEMANTIC_CLAUDE_READY_R01.zip') as z:
        names=[n for n in z.namelist() if n.startswith(prefix) and not n.endswith('/')]
        expected={n[len(prefix):] for n in names}
        actual={p.relative_to(base).as_posix() for p in base.rglob('*') if p.is_file()}
        require(expected==actual and len(expected)==35,'expanded reviewed input file set differs')
        for n in names:
            p=base/n[len(prefix):]
            require(not p.is_symlink() and p.read_bytes()==z.read(n),'expanded prior bytes differ: '+n)
    spec=importlib.util.spec_from_file_location('accepted_action1_r01',base/'source/verify_semantic_equivalence.py')
    module=importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
    authority={n:(base/'input/v4_authority'/n).read_bytes() for n in module.AUTH_PINS}
    candidate=(base/'input/VPL_V5_R08_HF03_T01_R01.zip').read_bytes()
    rules=(base/'rules/INVARIANT_MAPPINGS_R01.json').read_bytes()
    receipt=module.evaluate(candidate,authority,rules)
    require(sha(module.canonical(receipt))==BASE_RECEIPT_SHA,'baseline mechanical receipt differs')
    require(receipt['resolved']==0 and receipt['unresolved']==328,'baseline was adjudicated')
    return module, module.read_archive(candidate), authority, load_json(rules), receipt

def read_new_inputs(root: Path):
    pins=load_json((root/'INPUT_BINDINGS_R03.json').read_bytes())
    require(set(pins)=={'OWNER_PERSISTENCE_AMENDMENT_R03.md','OWNER_DECISION_RECEIPT_R03.json','originals/EXACT_CLOSEOUT_CORRECTIONS_R02.json','originals/OWNER_PERSISTENCE_DECISION_DRAFT_R02.md','originals/CLAUDE_ACTION1_CONFIRMATION_REPORT_R01.md','originals/CLAUDE_ACTION1_CONFIRMATION_R01.zip','originals/VPL_ACTION1_SEMANTIC_CLAUDE_READY_R01.zip'},'unexpected input binding set')
    for name,pin in pins.items():
        p=root/name
        require(not p.is_symlink() and sha(p.read_bytes())==pin,'new input hash mismatch: '+name)
    amendment=(root/'OWNER_PERSISTENCE_AMENDMENT_R03.md').read_bytes()
    owner=load_json((root/'OWNER_DECISION_RECEIPT_R03.json').read_bytes())
    require(owner.get('status')=='ADOPTED_DESIGN_ONLY','owner decision not adopted')
    require(owner.get('owner_reply_verbatim')=='approve' and owner.get('approved_proposition_verbatim')==APPROVAL,'owner proposition differs')
    require(owner.get('amendment_sha256')==sha(amendment),'owner amendment binding differs')
    require(owner.get('draft_sha256')==sha((root/'originals/OWNER_PERSISTENCE_DECISION_DRAFT_R02.md').read_bytes()),'wrong approved draft')
    for k in ['semantic_model_approved','r2_m01_closed','runtime_current_certified','scores_changed','deployment_authorized']:
        require(owner.get(k) is False,'approval scope widened: '+k)
    patches=load_json((root/'originals/EXACT_CLOSEOUT_CORRECTIONS_R02.json').read_bytes())
    require(patches.get('candidate_git_tree')=='ddc0e4fd644b59a29d1bb7a7f647cb89f2556f4d','wrong patch baseline')
    return amendment,owner,patches

def apply_context(data, entries, amendment, patches):
    require(len(entries)==35 and len({e['path'] for e in entries})==35,'source coverage incomplete')
    effective={e['path']:data[e['path']] for e in entries}
    ps=patches['patches']
    require(len(ps)==2,'unexpected patch count')
    require([(p['file'],p['array'],p['row_id'],p['field']) for p in ps]==[
        (GL,'rules','VPL-GL-15','requirement'),
        (AGENT,'agents','VPL-A01','historical_rev006_design.contract_authority')], 'unexpected patch scope')
    for p in ps:
        obj=load_json(effective[p['file']]); target=row(obj,p['array'],p['row_id'])
        require(canonical(target)==canonical(p['expected_row']),'patch precondition differs: '+p['row_id'])
        if p['row_id']=='VPL-GL-15':
            expected=('The V4 REV006 VPL-GL-15 requirement is retained in full and remains mandatory. The following V5 contract is additive, not a replacement; none of its wording waives an inherited V4 safeguard.\n\n'+V4_GL+'\n\nAdditional V5 contract:\n'+target['requirement'])
            require(p['replacement']==expected,'GL15 staged replacement lost complete inheritance')
            target['requirement']=expected
        else:
            require(target['historical_rev006_design']['contract_authority']==p['old'],'old A01 citation differs')
            require(p['replacement']=='build_framework_V4.0 Final_Rev6.html#persistence','incorrect A01 correction')
            target['historical_rev006_design']['contract_authority']=p['replacement']
        effective[p['file']]=pretty(obj)
    effective[O1]=data[O1]+'\n\n---\n\n## Later owner amendment — September 28, 2026 / R03\n\n'.encode()+amendment
    return effective

def assemble_bundle(effective, entries):
    header=("# VPL V5 — effective closeout context R03 (source-bound addendum)\n\n"
            "Base: frozen HF03/T01, not overwritten or promoted. Published general authority remains V4 FINAL / REV006.\n\n"
            "For this versioned V5 design closeout only, the later owner decision appended to Source 4 (O1) replaces the A01/M26 mandatory-when-proven 15-minute producer with invocation-bound proof and optional cron. It preserves every V4 GL-15 safeguard; V5 GL-15 is additive. Only the named design choice is newly adopted; no blanket approval of all candidate overlays is inferred. The same named producer choice applies to its repeated statements; all other inherited triggers and safeguards remain intact.\n\n"
            "All 35 base sources are included in original order. Source 4 has the appended actual owner decision; Source 26 has the Rev6 provenance correction; Source 27 has full V4 GL-15 retention. All other source bytes are unchanged. The complete original context and its historical claims remain available in the frozen archive. This is not an installable/resealed 451-file candidate.\n\n"
            "On an unresolved conflict: STOP for source/owner resolution. No runtime CURRENT, semantic model approval, R2-M01 closure or score is certified. The remaining independent review must address this complete context and the three existing registry wrapper differences.\n").encode()
    out=header
    for i,e in enumerate(entries,1):
        path=e['path']; raw=effective[path]
        out+=(f'\n\n## Source {i}: {path}\n\nSHA-256: {sha(raw)}\n\n<!-- BEGIN PINNED SOURCE -->\n').encode()+raw+b'\n<!-- END PINNED SOURCE -->\n'
    return out

def compose(root: Path):
    module,data,authority,old_rules,base=load_baseline(root)
    amendment,owner,patches=read_new_inputs(root)
    entries=base['v5_source_inventory']
    effective=apply_context(data,entries,amendment,patches)
    blocks,_=module.source_blocks(authority[module.V4_NAME],load_json(authority['VPL_V4_EQUAL_LOAD_MANIFEST_REV006.json']))
    require(row(load_json(blocks['V4_LOCKED_GUIDELINES.json']),'rules','VPL-GL-15')['requirement']==V4_GL,'V4 complete GL15 differs')
    inv=[{'path':e['path'],'base_sha256':e['sha256'],'sha256':sha(effective[e['path']]),'size_bytes':len(effective[e['path']]),'changed':effective[e['path']]!=data[e['path']]} for e in entries]
    require({r['path'] for r in inv if r['changed']}=={O1,GL,AGENT},'unexpected effective source change')
    bundle=assemble_bundle(effective,entries)
    v4=authority[module.V4_NAME]; require(bundle.count(v4)==1,'V4 not uniquely inherited')
    offset=bundle[:bundle.index(v4)].count(b'\n')
    maps=copy.deepcopy(old_rules['span_mappings'])
    # Inspect actual mapping shape rather than synthesizing a new 328-row inventory.
    for item in maps:
        dest=item['destination_span']; src=item['source_span']
        dest['file']='EFFECTIVE_CONTEXT_BUNDLE_R03.md'
        dest['start_line']=src['start_line']+offset; dest['end_line']=src['end_line']+offset
        left=b''.join(v4.splitlines(keepends=True)[src['start_line']-1:src['end_line']])
        right=b''.join(bundle.splitlines(keepends=True)[dest['start_line']-1:dest['end_line']])
        require(left==right and sha(left)==src['span_sha256'],'effective destination span differs')
    revised=copy.deepcopy(old_rules)
    revised['kind']='ACTION1_SOURCE_BOUND_RULE_MODEL_WITH_OWNER_ADDENDUM_R03'
    revised['approval_status']='OWNER_DESIGN_ADOPTED_SEMANTIC_ADEQUACY_PENDING'
    revised['base_rule_sha256']=sha((root/'prior_reviewed_R01/rules/INVARIANT_MAPPINGS_R01.json').read_bytes())
    revised['span_mappings']=maps
    revised['v5_source_inventory']=[{k:v for k,v in e.items() if k in ('path','sha256','size_bytes')} for e in inv]
    revised['base_registry_comparison']=revised.pop('registry_comparison')
    revised['base_bounded_translation_checks']=revised.pop('bounded_translation_checks')
    combined={**data,**effective}
    revised['registry_comparison']=module.registry_comparison(blocks,combined)
    revised['effective_context_bundle_sha256']=sha(bundle)
    revised['owner_design_binding']={'record_sha256':sha((root/'OWNER_DECISION_RECEIPT_R03.json').read_bytes()),'amendment_sha256':sha(amendment),'owner_reply':'approve','scope':'A01/M26_DESIGN_AND_FULL_GL15_ADDITIVE_RETENTION_ONLY','independent_semantic_approval':False}
    revised['rule_id']='SOURCE_BOUND_VERBATIM_INHERITANCE_WITH_EXPLICIT_OVERLAY_REVIEW_R03'
    revised['rule_definition']['r03_precedence']='Apply only the explicitly adopted A01/M26 design change, retain GL15 completely and add V5 wording. Other inherited triggers remain; aggregate overlay adequacy still requires independent review.'
    gl=row(load_json(effective[GL]),'rules','VPL-GL-15')['requirement']
    revised['added_gl15_literal_checks']=[{'literal':c,'present_in_effective_GL15':c in gl,'proof':'literal retention, not behavioral entailment'} for c in CLAUSES]
    revised['complete_v4_gl15_retained']=V4_GL in gl
    revised['complete_old_v5_gl15_retained']=row(load_json(data[GL]),'rules','VPL-GL-15')['requirement'] in gl
    revised['o1_prior_bytes_preserved']=effective[O1].startswith(data[O1])
    rules_raw=pretty(revised)
    receipt={'kind':'ACTION1_OWNER_CLOSEOUT_MECHANICAL_RECEIPT_R03','result':'CORRECTIONS_APPLIED_SEMANTIC_ADEQUACY_PENDING','owner_design_decision':'ADOPTED','base_mechanical_receipt_sha256':sha(module.canonical(base)),'candidate_archive_sha256':module.CANDIDATE_SHA,'candidate_git_tree':module.CANDIDATE_TREE,'candidate_files':451,'frozen_candidate_unchanged':True,'effective_sources':35,'unchanged_effective_sources':32,'changed_effective_sources':[e for e in inv if e['changed']],'mechanically_verified_spans':328,'semantic_resolved':0,'semantic_unresolved':328,'r2_m01_closed':False,'independent_semantic_approval_supplied':False,'full_v4_gl15_retained':revised['complete_v4_gl15_retained'],'full_old_v5_gl15_retained':revised['complete_old_v5_gl15_retained'],'missing_gl15_literals_now_present':all(c in gl for c in CLAUSES),'a01_reference_corrected':row(load_json(effective[AGENT]),'agents','VPL-A01')['historical_rev006_design']['contract_authority'],'owner_amendment_sha256':sha(amendment),'effective_context_bundle_sha256':sha(bundle),'revised_rules_sha256':sha(rules_raw),'frozen_candidate_source_code_changed':False,'candidate_suite_executed':False,'old_verifier_test_suite_rerun':False,'actions3_4_rerun':False,'scores_changed':False,'deployments_performed':False,'execution_boundary':'New composition and literal-preservation checks plus accepted baseline source verification only; no semantic adjudication.'}
    outputs={'effective_context/'+n:b for n,b in effective.items()}
    outputs.update({'EFFECTIVE_CONTEXT_BUNDLE_R03.md':bundle,'SEMANTIC_RULES_R03.json':rules_raw,'EFFECTIVE_CONTEXT_INVENTORY_R03.json':pretty(inv),'OWNER_CLOSEOUT_MECHANICAL_RECEIPT_R03.json':pretty(receipt)})
    return outputs, receipt

def write_new(out:Path, outputs:dict):
    require(not out.exists(),'output directory exists; refusing overwrite')
    require(out.parent.is_dir(),'parent output directory missing')
    out.mkdir()
    for n,b in outputs.items():
        p=out/n;p.parent.mkdir(parents=True,exist_ok=True)
        with p.open('xb') as f:f.write(b)

def main():
    ap=argparse.ArgumentParser(description=__doc__);ap.add_argument('--root',type=Path,default=Path(__file__).resolve().parents[1]);ap.add_argument('--output',type=Path,required=True)
    a=ap.parse_args()
    try:
        require(not a.output.exists(),'output directory exists; refusing overwrite')
        out,r=compose(a.root.resolve());write_new(a.output,out)
        print(json.dumps({'result':r['result'],'sources':35,'changed_sources':3,'spans':328,'semantic_resolved':0,'semantic_unresolved':328,'files_written':len(out),'receipt_sha256':sha(out['OWNER_CLOSEOUT_MECHANICAL_RECEIPT_R03.json'])},sort_keys=True));return 0
    except (Rejected,ValueError,OSError,KeyError,TypeError,zipfile.BadZipFile) as e:
        print(json.dumps({'result':'REJECTED','error':str(e)},sort_keys=True),file=sys.stderr);return 2
if __name__=='__main__':sys.exit(main())

7,827 bytes · SHA-256 b60c542bc0927cec689d347f0f3fa6042953af85dfca4bebeb0970d2d3dbfe8f

Download exact source file

#!/usr/bin/env python3
"""Read-only verification of generated R03 files; does not import the composer."""
from __future__ import annotations
import argparse,copy,hashlib,io,json,sys,zipfile
from pathlib import Path

CANDIDATE='c3bd53d775752494d982183cc71b551af6b4bf2c879416d6866a859e64fd28ac'
V4='d4e75170b791b4458d60ea39f35fa165a81b89fce236c12f15f40af038a95d62'
GL='controls/VPL_V5_06_GUIDELINE_REGISTRY_R02.json'
AG='controls/VPL_V5_06_AGENT_REGISTRY_R02.json'
O1='O1_ADOPTED_AMENDMENTS.md'
def sha(b):return hashlib.sha256(b).hexdigest()
def must(v,s):
    if not v:raise ValueError(s)
def j(b):return json.loads(b)
def getrow(d,a,id):
    rows=[x for x in d[a] if x['id']==id];must(len(rows)==1,'row missing/duplicate');return rows[0]

def verify(root:Path,generated:Path):
    b=root/'prior_reviewed_R01';raw=(b/'input/VPL_V5_R08_HF03_T01_R01.zip').read_bytes()
    must(sha(raw)==CANDIDATE,'wrong frozen candidate')
    with zipfile.ZipFile(io.BytesIO(raw)) as z:
        d={i.filename.split('/',1)[1]:z.read(i) for i in z.infolist() if not i.is_dir()}
    must(len(d)==451,'candidate count differs')
    entries=j(d['SEMANTIC_INPUTS_R08.json'])['files'];must(len(entries)==35,'expected 35 source contexts')
    inv=j((generated/'EFFECTIVE_CONTEXT_INVENTORY_R03.json').read_bytes())
    must([x['path'] for x in inv]==[x['path'] for x in entries],'context source order differs')
    must({p.relative_to(generated/'effective_context').as_posix() for p in (generated/'effective_context').rglob('*') if p.is_file()}=={e['path'] for e in entries},'missing/extra effective file')
    owner=(root/'OWNER_PERSISTENCE_AMENDMENT_R03.md').read_bytes()
    decision=j((root/'OWNER_DECISION_RECEIPT_R03.json').read_bytes())
    must(decision['owner_reply_verbatim']=='approve' and decision['status']=='ADOPTED_DESIGN_ONLY','no owner design decision')
    must(decision['amendment_sha256']==sha(owner),'decision binding failed')
    v4=(b/'input/v4_authority/VPL_V4_SEMANTIC_LOAD_BUNDLE_REV006.md').read_bytes();must(sha(v4)==V4,'wrong V4 source')
    manifest=j((b/'input/v4_authority/VPL_V4_EQUAL_LOAD_MANIFEST_REV006.json').read_bytes())
    slot=next(x for x in manifest['canonical_sources'] if x['name']=='V4_LOCKED_GUIDELINES.json')
    start=v4.index(b'# BEGIN CANONICAL SOURCE: V4_LOCKED_GUIDELINES.json\n\n')+len(b'# BEGIN CANONICAL SOURCE: V4_LOCKED_GUIDELINES.json\n\n')
    guidelines=v4[start:start+slot['bytes']];must(sha(guidelines)==slot['sha256'],'V4 guideline block mismatch')
    v4req=getrow(j(guidelines),'rules','VPL-GL-15')['requirement']
    oldreq=getrow(j(d[GL]),'rules','VPL-GL-15')['requirement']
    fullreq=('The V4 REV006 VPL-GL-15 requirement is retained in full and remains mandatory. The following V5 contract is additive, not a replacement; none of its wording waives an inherited V4 safeguard.\n\n'+v4req+'\n\nAdditional V5 contract:\n'+oldreq)
    eff={};changed=[]
    for old,new in zip(entries,inv):
        n=old['path'];v=(generated/'effective_context'/n).read_bytes();eff[n]=v
        must(new['sha256']==sha(v) and new['size_bytes']==len(v) and new['base_sha256']==sha(d[n]),'inventory binding failed: '+n)
        actual_change=v!=d[n];must(new['changed']==actual_change,'false change classification')
        if actual_change:changed.append(n)
        if n==GL:
            expected=j(d[n]);getrow(expected,'rules','VPL-GL-15')['requirement']=fullreq
            must(j(v)==expected,'GL15 changed beyond exact additive correction')
        elif n==AG:
            expected=j(d[n]);getrow(expected,'agents','VPL-A01')['historical_rev006_design']['contract_authority']='build_framework_V4.0 Final_Rev6.html#persistence'
            must(j(v)==expected,'A01 changed beyond exact citation correction')
        elif n==O1:
            must(v==d[n]+'\n\n---\n\n## Later owner amendment — September 28, 2026 / R03\n\n'.encode()+owner,'O1 original not preserved with exact appended decision')
        else:must(v==d[n],'unrelated source changed: '+n)
    must(set(changed)=={GL,AG,O1},'unexpected changed set')
    bundle=(generated/'EFFECTIVE_CONTEXT_BUNDLE_R03.md').read_bytes()
    header=bundle.split(b'\n\n## Source 1: ',1)[0]
    must(b'Only the named design choice is newly adopted' in header and b'On an unresolved conflict: STOP' in header,'precedence limitation missing')
    reconstructed=header
    for i,e in enumerate(entries,1):
        n=e['path'];v=eff[n]
        reconstructed+=(f'\n\n## Source {i}: {n}\n\nSHA-256: {sha(v)}\n\n<!-- BEGIN PINNED SOURCE -->\n').encode()+v+b'\n<!-- END PINNED SOURCE -->\n'
    must(reconstructed==bundle,'effective context cannot be reconstructed')
    rules_raw=(generated/'SEMANTIC_RULES_R03.json').read_bytes();rules=j(rules_raw)
    base_rules=j((b/'rules/INVARIANT_MAPPINGS_R01.json').read_bytes())
    must(rules['approval_status']=='OWNER_DESIGN_ADOPTED_SEMANTIC_ADEQUACY_PENDING','false semantic approval')
    must(len(rules['span_mappings'])==328,'span inventory incomplete')
    vl=v4.splitlines(keepends=True);bl=bundle.splitlines(keepends=True)
    must(bundle.count(v4)==1,'V4 not unique in effective bundle')
    offset=bundle[:bundle.index(v4)].count(b'\n')
    for before,after in zip(base_rules['span_mappings'],rules['span_mappings']):
        old=copy.deepcopy(before);new=copy.deepcopy(after);a=old.pop('destination_span');s=new.pop('destination_span')
        must(old==new,'non-destination mapping data changed')
        src=after['source_span']
        must(s=={**a,'file':'EFFECTIVE_CONTEXT_BUNDLE_R03.md','start_line':src['start_line']+offset,'end_line':src['end_line']+offset},'bad updated range')
        x=b''.join(vl[src['start_line']-1:src['end_line']]);y=b''.join(bl[s['start_line']-1:s['end_line']])
        must(x==y and sha(x)==src['span_sha256'],'span not preserved')
    r=j((generated/'OWNER_CLOSEOUT_MECHANICAL_RECEIPT_R03.json').read_bytes())
    must(r['semantic_resolved']==0 and r['semantic_unresolved']==328 and r['r2_m01_closed'] is False,'unearned closure')
    must(r['effective_context_bundle_sha256']==sha(bundle) and r['revised_rules_sha256']==sha(rules_raw),'receipt source binding mismatch')
    must(rules['effective_context_bundle_sha256']==sha(bundle),'rules context binding mismatch')
    must(rules['owner_design_binding']['record_sha256']==sha((root/'OWNER_DECISION_RECEIPT_R03.json').read_bytes()),'rules owner binding mismatch')
    for literal in ['Treat model persistence as untrusted.','immediate behavioral test','selective retrieval is not evidence']:
        must(literal in fullreq,'retained invariant missing')
    return {'kind':'OWNER_CLOSEOUT_SAVED_DATA_CHECK_R03','result':'PASS_SAVED_DATA_ONLY','original_candidate_files':451,'complete_context_sources':35,'unchanged_sources':32,'changed_sources':changed,'source_spans':328,'full_V4_GL15_and_full_old_V5_GL15_preserved':True,'three_missing_clauses_retained':True,'owner_design_adopted':True,'semantic_approval':False,'semantic_resolved':0,'semantic_unresolved':328,'r2_m01_closed':False,'context_bundle_sha256':sha(bundle),'rules_sha256':sha(rules_raw),'composer_imported':False,'independent_review':False}

def main():
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--root',type=Path,default=Path(__file__).resolve().parents[1]);p.add_argument('--generated',type=Path,required=True);p.add_argument('--output',type=Path)
    a=p.parse_args()
    try:
        if a.output:must(not a.output.exists(),'refusing overwrite')
        result=verify(a.root,a.generated);raw=(json.dumps(result,sort_keys=True,indent=2)+'\n').encode()
        if a.output:
            with a.output.open('xb') as f:f.write(raw)
        print(raw.decode());return 0
    except (OSError,ValueError,KeyError,TypeError) as e:print(json.dumps({'result':'REJECTED','error':str(e)}));return 2
if __name__=='__main__':sys.exit(main())

31,585 bytes · SHA-256 442d271ee56dade4fe6c4f7e7a002363a0471b1b5bc20b583d9b16d927280547

Download exact source file

#!/usr/bin/env python3
"""Source-bound Action 1 verifier. No candidate imports, execution, or source writes.

A successful mechanical run does NOT assert semantic equivalence. Unless an
externally approved, hash-pinned model adjudication is supplied, exit 3 and keep
semantic statuses UNRESOLVED. Exit 2 is malformed/untrusted input. Exit 0 is only
bounded specification equivalence under that declared approval, never runtime.
"""
from __future__ import annotations
import argparse
import hashlib
import io
import json
import re
import stat
import sys
import zipfile
from pathlib import Path, PurePosixPath
from typing import Any

CANDIDATE_SHA = 'c3bd53d775752494d982183cc71b551af6b4bf2c879416d6866a859e64fd28ac'
CANDIDATE_TREE = 'ddc0e4fd644b59a29d1bb7a7f647cb89f2556f4d'
ROOT_NAME = 'VPL_V5_R08_HF03_T01_R01'
V4_NAME = 'VPL_V4_SEMANTIC_LOAD_BUNDLE_REV006.md'
V4_SHA = 'd4e75170b791b4458d60ea39f35fa165a81b89fce236c12f15f40af038a95d62'
V5_NAME = 'VPL_V5_SEMANTIC_LOAD_BUNDLE_R08.md'
MAP = 'controls/V4_PRESERVATION_MAP_HF02.json'
MATRIX = 'controls/MANUAL_LIGHT_PRESERVATION_MATRIX_R02.md'
AUTH_PINS = {
    V4_NAME: V4_SHA,
    'VPL_V4_EQUAL_LOAD_MANIFEST_REV006.json': '1e7ce7be27906b1f3251735a1f38c3c14b7d34e5ded7d4679938c5e68d242aea',
    'PERSISTENCE_CHALLENGE_MATRIX_REV006.json': 'f3e224d4c18a5aa6f2efe3708a9d62607660965d3db446078e7c02a3f10e7ae9',
    'REV006_SHA256SUMS.txt': 'ccf8bce20b37c6f6613abc4c855734332ea7f37edff6f3ced57e590b08e32c98',
}
REGISTRIES = (
    ('AGENT_REGISTRY.json', 'controls/VPL_V5_06_AGENT_REGISTRY_R02.json', 'agents'),
    ('V4_MODULE_REGISTRY_FINAL.json', 'controls/VPL_V5_06_MODULE_REGISTRY_R02.json', 'modules'),
    ('V4_LOCKED_GUIDELINES.json', 'controls/VPL_V5_06_GUIDELINE_REGISTRY_R02.json', 'rules'),
)
MODEL_RULE = 'SOURCE_BOUND_VERBATIM_INHERITANCE_WITH_EXPLICIT_OVERLAY_REVIEW_R01'

class InvalidInput(ValueError):
    pass

def require(condition: bool, message: str) -> None:
    if not condition:
        raise InvalidInput(message)

def sha(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()

def canonical(x: Any) -> bytes:
    return (json.dumps(x, sort_keys=True, ensure_ascii=False, allow_nan=False,
                       separators=(',', ':')) + '\n').encode('utf-8')

def load_json(b: bytes) -> Any:
    def pairs(items):
        out = {}
        for k, v in items:
            require(k not in out, 'duplicate JSON key: ' + k)
            out[k] = v
        return out
    def bad_constant(value):
        raise InvalidInput('non-finite JSON value: ' + value)
    try:
        return json.loads(b.decode('utf-8'), object_pairs_hook=pairs,
                          parse_constant=bad_constant)
    except (UnicodeError, json.JSONDecodeError) as e:
        raise InvalidInput('invalid UTF-8 JSON') from e

def same(a: Any, b: Any) -> bool:
    return canonical(a) == canonical(b)

def valid_path(name: str) -> str:
    p = PurePosixPath(name)
    require(bool(name) and not name.startswith('/') and '\\' not in name and
            ':' not in name and '\x00' not in name and all(x not in ('', '.', '..')
            for x in name.split('/')), 'unsafe path: ' + repr(name))
    require(p.as_posix() == name, 'noncanonical path')
    return name

def read_archive(raw: bytes) -> dict[str, bytes]:
    """In-memory reads only; reject duplicates, traversal and nonregular members."""
    require(len(raw) <= 25_000_000, 'archive exceeds bounded input size')
    try:
        with zipfile.ZipFile(io.BytesIO(raw)) as z:
            require(len(z.infolist()) <= 2000, 'too many archive members')
            seen, data, total = set(), {}, 0
            for item in z.infolist():
                name = item.filename.rstrip('/') if item.is_dir() else item.filename
                valid_path(name)
                require(item.filename not in seen, 'duplicate archive member')
                seen.add(item.filename)
                mode = item.external_attr >> 16
                require(not stat.S_ISLNK(mode), 'archive symlink forbidden')
                require(not mode or stat.S_IFMT(mode) in (0, stat.S_IFREG, stat.S_IFDIR),
                        'nonregular archive member')
                require(not (item.flag_bits & 1), 'encrypted archive forbidden')
                if item.is_dir():
                    continue
                require(name.startswith(ROOT_NAME + '/'), 'wrong candidate root')
                total += item.file_size
                require(total <= 75_000_000 and item.file_size <= 10_000_000,
                        'uncompressed size exceeds bound')
                rel = name[len(ROOT_NAME) + 1:]
                data[valid_path(rel)] = z.read(item)
            return data
    except (zipfile.BadZipFile, RuntimeError, EOFError) as e:
        raise InvalidInput('invalid candidate archive') from e

def git_tree(data: dict[str, bytes]) -> str:
    tree: dict[str, Any] = {}
    for name, b in data.items():
        parts = valid_path(name).split('/')
        here = tree
        for part in parts[:-1]:
            require(not isinstance(here.get(part), bytes), 'path/file collision')
            here = here.setdefault(part, {})
        require(parts[-1] not in here, 'duplicate tree leaf')
        here[parts[-1]] = b
    def obj(kind: str, b: bytes) -> bytes:
        return hashlib.sha1(kind.encode() + b' ' + str(len(b)).encode() + b'\0' + b).digest()
    def digest(node) -> bytes:
        out = b''
        for name in sorted(node, key=lambda n: (n + ('/' if isinstance(node[n], dict) else '')).encode()):
            v = node[name]
            out += (b'40000' if isinstance(v, dict) else b'100644') + b' ' + name.encode() + b'\0'
            out += digest(v) if isinstance(v, dict) else obj('blob', v)
        return obj('tree', out)
    return digest(tree).hex()

def source_blocks(v4: bytes, manifest: dict) -> tuple[dict[str, bytes], list[dict]]:
    """Recover exact original files from named byte-counted V4 transport blocks."""
    require(manifest.get('canonical_source_count') == 29, 'authority must bind 29 sources')
    require(manifest.get('semantic_bundle_sha256') == sha(v4), 'manifest/V4 hash mismatch')
    require(manifest.get('semantic_bundle_bytes') == len(v4), 'manifest/V4 length mismatch')
    declared = manifest.get('canonical_sources')
    require(isinstance(declared, list) and len(declared) == 29, 'incomplete canonical source list')
    blocks, observations = {}, []
    begins = re.findall(rb'^# BEGIN CANONICAL SOURCE: (.+)$', v4, re.M)
    require(len(begins) == 29 and len(set(begins)) == 29, 'missing/duplicate source blocks')
    require([x.decode() for x in begins] == [d['name'] for d in declared], 'source order differs')
    manifest_lines = re.findall(rb'^- `([^`]+)` \xe2\x80\x94 ([\d,]+) bytes \xe2\x80\x94 `([a-f0-9]{64})`$', v4, re.M)
    observed_manifest = [{'name': n.decode(), 'bytes': int(s.replace(b',', b'')),
                          'sha256': h.decode()} for n, s, h in manifest_lines]
    require(same(observed_manifest, declared), 'embedded and recovered manifest differ')
    for row in declared:
        name = valid_path(row['name'])
        require(name not in blocks and type(row['bytes']) is int and row['bytes'] > 0,
                'invalid canonical source row')
        marker = b'# BEGIN CANONICAL SOURCE: ' + name.encode() + b'\n\n'
        require(v4.count(marker) == 1, 'ambiguous canonical source marker')
        start = v4.index(marker) + len(marker)
        raw = v4[start:start + row['bytes']]
        endmarker = b'\n\n# END CANONICAL SOURCE: ' + name.encode() + b'\n'
        require(v4[start + len(raw):].startswith(endmarker), 'canonical source terminator differs')
        require(sha(raw) == row['sha256'], 'canonical source bytes differ: ' + name)
        blocks[name] = raw
        observations.append({**row, 'bundle_start_byte': start, 'hash_verified': True})
    return blocks, observations

def derive_spans(v4: bytes) -> list[dict]:
    """Re-derive the frozen inventory's heading spans, including its preambles.

The original inventory treats hash-prefixed headings inside fenced examples as
section boundaries. Preserve that exact scope; do not relabel these as 328 clauses.
"""
    lines = v4.decode('utf-8').splitlines(keepends=True)
    result, src, start, heading = [], None, None, None
    for pos, line in enumerate(lines):
        if line.startswith('# BEGIN CANONICAL SOURCE: '):
            require(src is None, 'nested canonical block')
            src, start, heading = line.strip().split(': ', 1)[1], pos + 1, 'SOURCE_PREAMBLE'
        elif src is not None and (line.startswith('# END CANONICAL SOURCE: ') or
                                   re.match(r'^#{1,6} ', line)):
            if any(x.strip() for x in lines[start:pos]):
                raw = ''.join(lines[start:pos]).encode()
                entry = {'source': src, 'heading': heading, 'start_line': start + 1,
                         'end_line': pos, 'span_sha256': sha(raw)}
                entry['span_id'] = 'S-' + sha(canonical(entry))[:24]
                result.append(entry)
            if line.startswith('# END CANONICAL SOURCE: '):
                require(line.strip().split(': ', 1)[1] == src, 'wrong source end marker')
                src, start, heading = None, None, None
            else:
                start, heading = pos, line.strip()
    require(src is None and len(result) == 328, 'expected complete 328-span inventory')
    require(len({r['span_id'] for r in result}) == 328, 'duplicate derived span identity')
    return result

def verify_source_map(mapping: dict, spans: list[dict], v4: bytes) -> None:
    require(mapping.get('source_sha256') == sha(v4), 'preservation map V4 pin differs')
    rows = mapping.get('source_sections')
    require(isinstance(rows, list) and len(rows) == 328, 'source map incomplete')
    keys = ('source', 'heading', 'start_line', 'end_line', 'span_sha256')
    for row, derived in zip(rows, spans):
        require(same({k: row.get(k) for k in keys}, {k: derived[k] for k in keys}),
                'map coverage/range/identity differs')
        require(row.get('destination') == V5_NAME and row.get('disposition') == 'RETAIN',
                'map destination/disposition differs')
    # semantic_equivalence/status/resolved metadata is deliberately NOT consulted.

def verify_v5_bundle(data: dict[str, bytes], v4: bytes) -> tuple[int, list[dict]]:
    idx = load_json(data['SEMANTIC_INPUTS_R08.json'])
    entries = idx.get('files')
    require(isinstance(entries, list) and len(entries) == 35, 'V5 input inventory incomplete')
    expected = ['frozen-source/' + V4_NAME, 'VPL_V5_EXECUTION_BIBLE_R10.md',
                'VPL_V5_REMEDIATION_BIBLE_R05.md', 'O1_ADOPTED_AMENDMENTS.md']
    expected += sorted(n for n in data if n.startswith('controls/') and n.count('/') == 1
                       and PurePosixPath(n).suffix in ('.json', '.md'))
    expected += ['templates/BUILD_RECAP_STRUCTURE_R02.html']
    require([e['path'] for e in entries] == expected, 'V5 source set/order differs')
    require(idx.get('no_selective_pruning') is True and idx.get('runtime_ingestion_certified') is False,
            'invalid semantic source scope')
    require(idx.get('module_ids') == [f'M{i:02}' for i in range(1, 27)], 'module scope differs')
    bundle = data[V5_NAME]
    first = b'\n\n## Source 1: '
    require(bundle.count(first) == 1, 'ambiguous first V5 source')
    header = bundle.split(first, 1)[0]
    require(b'Published authority remains V4 FINAL / REV006.' in header and
            b'On an unresolved conflict: STOP for source/owner resolution.' in header,
            'V5 authority/conflict boundary missing')
    rebuilt = header
    for i, e in enumerate(entries, 1):
        path = valid_path(e['path'])
        require(path in data and sha(data[path]) == e['sha256'] and len(data[path]) == e['size_bytes'],
                'V5 semantic entry hash/size differs: ' + path)
        rebuilt += (f'\n\n## Source {i}: {path}\n\nSHA-256: {e["sha256"]}\n\n'
                    '<!-- BEGIN PINNED SOURCE -->\n').encode() + data[path] + b'\n<!-- END PINNED SOURCE -->\n'
    require(rebuilt == bundle, 'V5 bundle cannot be reconstructed from complete source list')
    require(data['frozen-source/' + V4_NAME] == v4 and bundle.count(v4) == 1,
            'recovered V4 authority not uniquely verbatim in V5')
    return bundle[:bundle.index(v4)].count(b'\n'), entries

def field_diffs(old: dict, new: dict) -> list[dict]:
    """Keep missing and JSON null distinct, and preserve complete values."""
    return [{'field': k, 'source_present': k in old, 'destination_present': k in new,
             'source_value': old.get(k), 'destination_value': new.get(k)}
            for k in sorted(set(old) | set(new))
            if k not in old or k not in new or not same(old[k], new[k])]

def registry_comparison(blocks: dict, data: dict) -> dict:
    rows, wrappers = [], []
    for source, dest, array in REGISTRIES:
        a, b = load_json(blocks[source]), load_json(data[dest])
        def keyed(x):
            require(isinstance(x.get(array), list) and len(x[array]) > 0, 'missing registry rows')
            out = {}
            for r in x[array]:
                require(isinstance(r, dict) and isinstance(r.get('id'), str), 'invalid registry row')
                require(r['id'] not in out, 'duplicate registry ID')
                out[r['id']] = r
            return out
        aa, bb = keyed(a), keyed(b)
        require(set(aa) == set(bb), 'registry IDs added/removed without mapping')
        for rid in sorted(aa):
            diffs = field_diffs(aa[rid], bb[rid])
            rows.append({'id': rid, 'source_file': source, 'destination_file': dest,
                         'array': array, 'source_value': aa[rid], 'destination_value': bb[rid],
                         'source_payload_sha256': sha(canonical(aa[rid])),
                         'destination_payload_sha256': sha(canonical(bb[rid])),
                         'comparison': 'NONIDENTICAL_REQUIRES_RULE' if diffs else 'IDENTICAL_ROW_VALUE',
                         'field_differences': diffs})
        wrappers.append({'source_file': source, 'destination_file': dest,
                         'field_differences': field_diffs({k: v for k, v in a.items() if k != array},
                                                        {k: v for k, v in b.items() if k != array}),
                         'interpretation': 'V4 wrapper remains inherited; V5 wrapper is not an identical replacement. Model approval required.'})
    return {'rows': rows, 'wrapper_context_differences': wrappers,
            'total_registry_rows': len(rows),
            'identical_row_values': sum(r['comparison'] == 'IDENTICAL_ROW_VALUE' for r in rows),
            'changed_row_values': sum(r['comparison'] != 'IDENTICAL_ROW_VALUE' for r in rows)}

def bounded_translation_checks(reg: dict, data: dict) -> dict:
    """Execute explicit PROPOSED source-to-destination predicate mappings.

    These predicates are not approved semantic rules merely because they pass.
    M01/M25 are exact transformations, not similarity tests. Persistence pairs
    make the human interpretation assumption visible with exact source anchors.
    Scheduler replacement is reported as a change, never an identity proof.
    """
    by_id = {r['id']: r for r in reg['rows']}
    checks = []
    m01 = by_id['M01']
    old, new = m01['source_value']['contract'], m01['destination_value']['contract']
    a = 'REV 006 retains'
    b = 'V4 FINAL / REV006 authority also requires'
    suffix = ' Candidate mutable position is read only from VPL_V5_CURRENT_STATE.json.'
    checks.append({'id': 'TR-M01', 'operator': 'EXACT_NAMED_SUBSTITUTION_AND_APPEND',
                   'source_id': 'M01', 'source_phrase': a, 'destination_phrase': b,
                   'append': suffix, 'predicate_pass': old.count(a) == 1 and new == old.replace(a,b) + suffix,
                   'semantic_adjudication': 'REQUIRED_NOT_INFERRED'})
    m25 = by_id['M25']
    suffix = ' A trigger while this module remains proposed must return UNCERTAIN_ESCALATE and STOP; owner activation is separate.'
    checks.append({'id': 'TR-M25', 'operator': 'EXACT_APPEND_WITH_ORIGINAL_CONTRACT_UNCHANGED',
                   'source_id': 'M25', 'append': suffix,
                   'predicate_pass': m25['destination_value']['contract'] == m25['source_value']['contract'] + suffix,
                   'semantic_adjudication': 'REQUIRED_NOT_INFERRED'})
    current = data['controls/PERSISTENCE_STATE_R02.md'].decode('utf-8')
    pairs = [
        ('M26', 'contract', 'Load all canonical V4 module semantics equally', 'loaded the full equal-priority semantic set', 'equal-priority full load'),
        ('M26', 'contract', 'balanced anti-parroting challenge plus behavioral proof', 'required balanced behavioral challenge', 'balanced challenge'),
        ('M26', 'contract', 'save complete receipt', 'banked a matching durable receipt with readback', 'durable source-bound receipt'),
        ('VPL-GL-15', 'requirement', 'Load every canonical V4 operating control with equal persistence priority', 'loaded the full equal-priority semantic set', 'equal-priority full load'),
        ('VPL-GL-15', 'requirement', 'balanced challenge coverage and immediate behavioral test', 'required balanced behavioral challenge', 'balanced challenge plus action-bound contract'),
        ('VPL-GL-15', 'requirement', 'Self-report, recap or selective retrieval is not evidence', 'a verbal recap, or a contract-test PASS alone cannot establish CURRENT', 'recap/test PASS insufficient'),
    ]
    for i, (rid, field, left, right, concept) in enumerate(pairs, 1):
        checks.append({'id': f'PR-{i:02}', 'operator': 'PROPOSED_LITERAL_ANCHORED_PROPOSITION_PAIR',
                       'concept': concept, 'source_id': rid, 'source_field': field,
                       'source_anchor': left, 'destination_file': 'controls/PERSISTENCE_STATE_R02.md',
                       'destination_anchor': right,
                       'predicate_pass': left in by_id[rid]['source_value'][field] and right in current,
                       'semantic_adjudication': 'PAIR_MEANING_REQUIRES_APPROVAL; matching anchors is not a general entailment proof'})
    a01, m26 = by_id['VPL-A01'], by_id['M26']
    scheduler_change = (a01['source_value'].get('schedule') != a01['destination_value'].get('schedule'))
    module_change = ('Hostinger 15-minute full-bundle producer' in m26['source_value']['contract'] and
                     'optional freshness/watchdog producer' in m26['destination_value']['contract'])
    return {'checks': checks, 'predicates_passed': sum(x['predicate_pass'] for x in checks),
            'predicates_total': len(checks),
            'cron_policy_difference_detected': scheduler_change and module_change,
            'cron_policy_classification': 'NONIDENTICAL_DESIGN_REQUIRES_EXPLICIT_OWNER_DISPOSITION',
            'rule_status': 'PROPOSED_FOR_SEMANTIC_ADJUDICATION_NOT_SELF_APPROVED'}


def verify_rules(rules: dict, spans: list[dict], offset: int, reg: dict,
                 matrix: bytes, entries: list[dict]) -> list[dict]:
    require(isinstance(rules, dict), 'rule model must be an object')
    require(rules.get('kind') == 'ACTION1_SOURCE_BOUND_RULE_MODEL_R01' and
            rules.get('rule_id') == MODEL_RULE, 'unknown rule model')
    require(rules.get('proof_scope') == 'BOUNDED_SPECIFICATION_MODEL_NOT_RUNTIME', 'scope overclaim')
    require(rules.get('v4_sha256') == V4_SHA and rules.get('candidate_sha256') == CANDIDATE_SHA,
            'rule source binding differs')
    require(rules.get('matrix_sha256') == sha(matrix), 'rule matrix binding differs')
    require(same(rules.get('v5_source_inventory'), entries), 'rule overlay inventory differs')
    require(same(rules.get('registry_comparison'), reg), 'rule registry values/context differ')
    maps = rules.get('span_mappings')
    require(isinstance(maps, list) and len(maps) == len(spans), 'rule mappings incomplete')
    out = []
    for expected, row in zip(spans, maps):
        require(isinstance(row, dict), 'span mapping must be an object')
        require(same(row.get('source_span'), expected), 'mapping source span altered or reordered')
        require(row.get('operator') == 'EXACT_SPAN_IN_COMPLETE_SOURCE_CONTEXT', 'unknown span operator')
        destination = {'file': V5_NAME, 'start_line': expected['start_line'] + offset,
                       'end_line': expected['end_line'] + offset, 'span_sha256': expected['span_sha256']}
        require(same(row.get('destination_span'), destination), 'mapping destination wrong')
        require(row.get('manual') == 'LOAD_ALL_PRESERVE_SOURCE_DEFINED_TRIGGERS' and
                row.get('light') == 'LOAD_ALL_PRESERVE_SOURCE_DEFINED_TRIGGERS', 'route/activation assumption changed')
        require(row.get('approval_requirement') == 'EXTERNAL_MODEL_ADJUDICATION', 'missing approval boundary')
        out.append({'span_id': expected['span_id'], 'source': expected,
                    'destination': destination, 'operator': row['operator'],
                    'manual': row['manual'], 'light': row['light'],
                    'mechanical_result': 'PASS_VERBATIM_SPECIFICATION_MAPPING',
                    'semantic_status': 'UNRESOLVED',
                    'reason': 'Exact-span/context predicate passes; this does not approve the inheritance/overlay semantic model.'})
    return out

def check_adjudication(raw: bytes | None, trusted_sha: str | None, rules_sha: str,
                       reg: dict) -> tuple[bool, dict]:
    """External trust input, not a signature, reviewer identity, or self-approval.

A future operator must supply BOTH an approval record and its independently
trusted SHA. This wave supplies no such record. Synthetic fixtures are tests only.
"""
    if raw is None:
        require(trusted_sha is None, 'approval hash without record')
        return False, {'state': 'NOT_SUPPLIED', 'identity_attested': False}
    require(trusted_sha is not None and bool(re.fullmatch('[0-9a-f]{64}', trusted_sha)) and
            sha(raw) == trusted_sha, 'untrusted approval record')
    a = load_json(raw)
    require(isinstance(a, dict), 'approval must be an object')
    require(a.get('kind') == 'ACTION1_EXTERNAL_SPECIFICATION_MODEL_ADJUDICATION_R01', 'wrong approval kind')
    require(a.get('rules_sha256') == rules_sha and a.get('candidate_sha256') == CANDIDATE_SHA and
            a.get('v4_sha256') == V4_SHA, 'approval binds different source/model')
    require(a.get('decision') == 'APPROVE_BOUNDED_SPECIFICATION_MODEL' and
            a.get('rule_id') == MODEL_RULE, 'no explicit semantic model approval')
    require(a.get('all_328_span_mappings_reviewed') is True and
            a.get('all_35_source_overlay_effects_reviewed') is True and
            a.get('wrapper_context_differences_reviewed') is True, 'incomplete interpretation review')
    for key in ('owner_decision_reference', 'independent_review_reference'):
        require(isinstance(a.get(key), str) and bool(a[key].strip()), 'missing approval provenance')
    ids = sorted(r['id'] for r in reg['rows'] if r['field_differences'])
    require(a.get('approved_changed_registry_ids') == ids, 'changed registry decisions incomplete')
    return True, {'state': 'SUPPLIED_HASH_PINNED_DECLARATION', 'sha256': sha(raw),
                  'owner_decision_reference': a['owner_decision_reference'],
                  'independent_review_reference': a['independent_review_reference'],
                  'identity_attested': False,
                  'trust_boundary': 'Supplied trusted hash and documentary declarations; authenticity/semantic judgment are external, not proved by this program.'}

def evaluate(candidate: bytes, authority: dict[str, bytes], rules_raw: bytes,
             adjudication: bytes | None = None, adjudication_sha: str | None = None) -> dict:
    require(sha(candidate) == CANDIDATE_SHA, 'candidate archive pin mismatch')
    data = read_archive(candidate)
    require(len(data) == 451 and git_tree(data) == CANDIDATE_TREE, 'candidate count/tree differs')
    for name, pin in AUTH_PINS.items():
        require(name in authority and sha(authority[name]) == pin, 'recovered authority pin mismatch: ' + name)
    v4 = authority[V4_NAME]
    manifest = load_json(authority['VPL_V4_EQUAL_LOAD_MANIFEST_REV006.json'])
    challenge = load_json(authority['PERSISTENCE_CHALLENGE_MATRIX_REV006.json'])
    sums = {line.split('  ', 1)[1]: line.split('  ', 1)[0]
            for line in authority['REV006_SHA256SUMS.txt'].decode().splitlines()}
    for name in (V4_NAME, 'VPL_V4_EQUAL_LOAD_MANIFEST_REV006.json', 'PERSISTENCE_CHALLENGE_MATRIX_REV006.json'):
        require(sums.get(name) == sha(authority[name]), 'recovered file fails original checksum record')
    blocks, block_proofs = source_blocks(v4, manifest)
    spans = derive_spans(v4)
    mapping = load_json(data[MAP])
    verify_source_map(mapping, spans, v4)
    offset, entries = verify_v5_bundle(data, v4)
    matrix = data[MATRIX]
    require(mapping.get('matrix_sha256') == sha(matrix), 'matrix/map hash mismatch')
    base_ids = re.findall(rb'^\| ((?:P|M|L|C)\d{2}) \|', matrix, re.M)
    h_ids = re.findall(rb'^\| (H\d{2}) \|', matrix, re.M)
    require(len(base_ids) == len(set(base_ids)) == 41 and h_ids == [f'H{i:02}'.encode() for i in range(1,8)], 'matrix rows missing/duplicated')
    reg = registry_comparison(blocks, data)
    rules = load_json(rules_raw)
    results = verify_rules(rules, spans, offset, reg, matrix, entries)
    translation = bounded_translation_checks(reg, data)
    require(same(rules.get('bounded_translation_checks'), translation), 'translation predicate mapping differs')
    require(translation['predicates_passed'] == translation['predicates_total'], 'bounded translation predicate failed')
    v4_lines = v4.splitlines(keepends=True)
    v5_lines = data[V5_NAME].splitlines(keepends=True)
    for result in results:
        a, b = result['source'], result['destination']
        aa = b''.join(v4_lines[a['start_line']-1:a['end_line']])
        bb = b''.join(v5_lines[b['start_line']-1:b['end_line']])
        require(aa == bb and sha(aa) == a['span_sha256'], 'actual mapped span differs')
    # Evaluate authoritative machine-readable persistence predicates, without
    # pretending a schema/challenge definition is an executed behavioral proof.
    require(manifest.get('no_priority_weighting') is True, 'equal-priority invariant missing')
    require(challenge.get('mutation_requirement') == 'PERSISTENCE_STATE=CURRENT', 'mutation invariant missing')
    require(challenge.get('proofs') == ['semantic_completeness', 'cross_reference_understanding',
                                      'false_premise_rejection', 'behavioral_follow_through'], 'proof requirement differs')
    approved, approval = check_adjudication(adjudication, adjudication_sha, sha(rules_raw), reg)
    if approved:
        for result in results:
            result['semantic_status'] = 'EQUIVALENT_UNDER_EXTERNALLY_APPROVED_BOUNDED_MODEL'
            result['reason'] = 'Derived exact-span/context predicate passed under the hash-pinned external model adjudication; no runtime claim.'
    missing = [name for name in ('VPL_V4_EQUAL_LOAD_MANIFEST_REV006.json', 'PERSISTENCE_CHALLENGE_MATRIX_REV006.json')
               if not any(PurePosixPath(p).name == name for p in data)]
    return {'kind': 'SEMANTIC_EQUIVALENCE_RECEIPT_R08_ACTION1_R01',
            'result': 'PASS_UNDER_EXTERNALLY_APPROVED_BOUNDED_MODEL' if approved else 'MECHANICAL_PASS_SEMANTIC_APPROVAL_PENDING',
            'exit_code': 0 if approved else 3,
            'total_spans': len(results), 'mechanically_verified_spans': len(results),
            'resolved': len(results) if approved else 0,
            'unresolved': 0 if approved else len(results),
            'r2_m01_closed': False, 'closure_note': 'Independent coordinator/owner closure is external; this tool never promotes the finding.',
            'proof_scope': 'BOUNDED_SPECIFICATION_MODEL_NOT_RUNTIME',
            'verifier_sha256': sha(Path(__file__).read_bytes()),
            'source_binding': {'candidate_archive_sha256': sha(candidate), 'candidate_git_tree': git_tree(data),
                               'candidate_files': len(data), 'v4_bundle_sha256': sha(v4),
                               'v5_bundle_sha256': sha(data[V5_NAME]), 'matrix_sha256': sha(matrix),
                               'rule_model_sha256': sha(rules_raw)},
            'source_blocks': block_proofs, 'v5_source_inventory': entries,
            'matrix_base_rows_verified': 41, 'matrix_h_rows_verified': 7,
            'registry_comparison': reg, 'bounded_translation_checks': translation, 'external_model_adjudication': approval,
            'recovered_authority_files': [{'file': n, 'sha256': sha(authority[n]),
                                          'original_checksum_matches': True} for n in AUTH_PINS if n != 'REV006_SHA256SUMS.txt'],
            'supplemented_original_candidate_omissions': missing,
            'supplement_note': 'Recovered V4 authority accompanies the sidecar; missing originals were NOT inserted into or backdated into frozen candidate.',
            'persistence_definition_checks': {'equal_priority': True, 'complete_29_sources': True,
                                            'balanced_four_proofs_required': True, 'runtime_proofs_executed': False},
            'results': results,
            'zero_manual_decisions_during_execution': True,
            'rule_authorship_and_semantic_adjudication_are_not_automated': True,
            'arbitrary_natural_language_conflicts_exhaustively_detected': False,
            'runtime_or_model_persistence_certified': False, 'candidate_code_executed': False,
            'source_written': False, 'score_change_established': False}

def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--candidate', type=Path, required=True)
    parser.add_argument('--v4-authority', type=Path, required=True)
    parser.add_argument('--rules', type=Path, required=True)
    parser.add_argument('--output', type=Path, required=True)
    parser.add_argument('--adjudication', type=Path)
    parser.add_argument('--adjudication-sha256')
    args = parser.parse_args(argv)
    try:
        require(not args.output.exists(), 'refusing to overwrite existing output')
        require(args.output.parent.is_dir(), 'output directory must already exist')
        authority = {n: (args.v4_authority/n).read_bytes() for n in AUTH_PINS}
        receipt = evaluate(args.candidate.read_bytes(), authority, args.rules.read_bytes(),
                           args.adjudication.read_bytes() if args.adjudication else None,
                           args.adjudication_sha256)
        with args.output.open('xb') as f:
            f.write(canonical(receipt))
        print(json.dumps({'result': receipt['result'], 'total_spans': receipt['total_spans'],
                          'mechanically_verified_spans': receipt['mechanically_verified_spans'],
                          'resolved': receipt['resolved'], 'unresolved': receipt['unresolved'],
                          'receipt_sha256': sha(args.output.read_bytes())}, sort_keys=True))
        return receipt['exit_code']
    except (InvalidInput, OSError, KeyError, TypeError, IndexError) as e:
        print(json.dumps({'result': 'INPUT_REJECTED', 'error': str(e)}, sort_keys=True), file=sys.stderr)
        return 2

if __name__ == '__main__':
    sys.exit(main())

14,677 bytes · SHA-256 92d436e1ebb17411186d764d81d6883daa35e455264b6096553641c39d45b861

Download exact source file

#!/usr/bin/env python3
"""Bounded T01 confirmation harness; only disposable output directories are written.
Uses real source preflight and runner execution, never journal-supplied executable code.
ACTION3 R01 integrates a test-only source-frame stall for --timeout-only. The fault
uses a real subprocess timeout; normal execution and final resume are uninjected.
No production/runtime certification or semantic-equivalence proof is implied.
"""
from __future__ import annotations
import argparse,copy,hashlib,importlib.util,json,math,os,runpy,shutil,subprocess,sys,time,types
sys.dont_write_bytecode = True
from pathlib import Path
from datetime import datetime,timezone

def sha(p):return hashlib.sha256(Path(p).read_bytes()).hexdigest()
def write(p,d):Path(p).write_text(json.dumps(d,indent=2)+'\n')

# ACTION3 R01: test-only fault injection, never a candidate-source patch.
# The real selected source frame is held before its first statement. The parent
# waits for that handshake before starting a real subprocess timeout. Startup
# speed and the runtime of unrelated commands therefore cannot select the fault.

def positive_seconds(value):
 value=float(value)
 if not math.isfinite(value) or value <= 0:
  raise argparse.ArgumentTypeError('timing must be finite and positive')
 return value

def nonnegative_seconds(value):
 value=float(value)
 if not math.isfinite(value) or value < 0:
  raise argparse.ArgumentTypeError('delay must be finite and nonnegative')
 return value

def wait_for_source_frame(proc,ready,target,limit):
 deadline=time.monotonic()+limit
 while not ready.exists():
  code=proc.poll()
  if code is not None:
   raise RuntimeError('fault child exited before source-frame handshake: '+str(code))
  if time.monotonic() >= deadline:
   raise RuntimeError('source-frame handshake deadline exceeded; NOT a verified timeout')
  time.sleep(0.01)
 event=json.loads(ready.read_text())
 if (event.get('kind')!='ACTION3_SOURCE_FRAME_STALL' or
     event.get('pid')!=proc.pid or event.get('target')!=str(target) or
     event.get('target_sha256')!=sha(target)):
  raise RuntimeError('invalid source-frame handshake')
 return event

def run_with_source_stall(command,ready,timeout_seconds,ready_seconds,startup_delay,**kwargs):
 """Run actual source through a test-only entry-frame gate; raise real timeout."""
 if not isinstance(command,list) or len(command)<3 or command[1]!='-B':
  raise ValueError('unexpected source command shape')
 target=Path(command[2]).resolve()
 if ready.exists():
  raise ValueError('refusing existing fault evidence')
 gated=[sys.executable,'-B',str(Path(__file__).resolve()),'--action3-gated-child',
        str(ready),str(startup_delay),str(target),*command[3:]]
 kwargs.pop('timeout',None)  # Start fault budget ONLY after the handshake.
 proc=subprocess.Popen(gated,**kwargs)
 try:
  wait_for_source_frame(proc,ready,target,ready_seconds)
  # This is the real subprocess wait, not a fabricated exception or exit code.
  code=proc.wait(timeout=timeout_seconds)
  return subprocess.CompletedProcess(command,code)
 finally:
  if proc.poll() is None:
   proc.kill()
  proc.wait()

def gated_child(argv):
 ready=Path(argv[0]);delay=nonnegative_seconds(argv[1]);target=Path(argv[2]).resolve()
 if ready.exists():
  raise ValueError('ready path already exists')
 time.sleep(delay)  # Optional explicit slow-start regression fixture.
 sys.argv=[str(target),*argv[3:]];sys.path.insert(0,str(target.parent))
 def at_source(frame,event,arg):
  if event=='call' and frame.f_code.co_filename==str(target):
   body={'kind':'ACTION3_SOURCE_FRAME_STALL','pid':os.getpid(),
         'target':str(target),'target_sha256':sha(target),
         'phase':'before_first_statement','test_only':True}
   temp=ready.with_suffix('.pending');write(temp,body);temp.replace(ready)
   print('ACTION3 TEST-ONLY STALL: '+str(target),flush=True)
   while True:time.sleep(3600)  # Parent kills this process on its bounded timeout.
  return at_source
 sys.settrace(at_source)
 try:runpy.run_path(str(target),run_name='__main__')
 finally:sys.settrace(None)
 raise RuntimeError('target frame was never intercepted')

def subprocess_proxy(run):
 return types.SimpleNamespace(**dict(vars(subprocess),run=run))

def fault_runner(argv):
 script=Path(argv[0]).resolve();index=int(argv[1]);ready=Path(argv[2]).resolve()
 budget=positive_seconds(argv[3]);startup=positive_seconds(argv[4]);delay=nonnegative_seconds(argv[5])
 sys.path.insert(0,str(script.parent))
 spec=importlib.util.spec_from_file_location('action3_frozen_runner',script)
 module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
 plan=module.command_plan()
 if not 1<=index<=len(plan):raise ValueError('fault index outside actual plan')
 target_name=plan[index-1][0]
 if sum(name==target_name for name,_ in plan)!=1:raise ValueError('ambiguous fault target')
 injected=[]
 def controlled_run(command,**kwargs):
  if len(command)>=3 and Path(command[2]).name==target_name:
   if injected:raise RuntimeError('duplicate injection attempt')
   injected.append(target_name)
   return run_with_source_stall(command,ready,budget,startup,delay,**kwargs)
  return subprocess.run(command,**kwargs)
 # Change only this imported runner's subprocess binding for this fault run.
 # Do not patch the global subprocess module, source files, plan, hashes or receipts.
 module.subprocess=subprocess_proxy(controlled_run)
 sys.argv=[str(script),*argv[6:]]
 code=module.main()
 if not injected:raise RuntimeError('requested fault was not reached')
 return code

def validate_timeout_checkpoint(seed,journal,failure,index):
 """Derived state checks: no completed work may be credited to a failed attempt."""
 checks=[
  (failure.get('result')=='FAIL','missing failed reproduction receipt'),
  (journal.get('commands')==seed.get('commands'),'verified checkpoint advanced or changed'),
  (len(journal.get('attempts',[]))==len(seed.get('attempts',[]))+1,'unexpected attempt count'),
  (journal.get('state')=='FAILED_OR_INCOMPLETE','journal not marked failed/incomplete'),
 ]
 for passed,reason in checks:
  if not passed:raise AssertionError(reason)
 attempt=journal['attempts'][-1]
 if (attempt.get('index')!=index or attempt.get('exit_code')!=124 or
     attempt.get('state')!='FAILED_OR_INCOMPLETE'):
  raise AssertionError('requested command did not record a genuine incomplete timeout')
 return attempt

def main():
 if sys.flags.optimize:raise RuntimeError('Run without -O / PYTHONOPTIMIZE; assertions must be enabled')
 p=argparse.ArgumentParser(description=__doc__);p.add_argument('--candidate',type=Path,required=True);p.add_argument('--archive',type=Path,required=True);p.add_argument('--out',type=Path,required=True);p.add_argument('--actor',required=True);p.add_argument('--session',required=True);p.add_argument('--timeout-only',action='store_true');p.add_argument('--fault-at',type=int,default=3);p.add_argument('--fault-timeout',type=positive_seconds,default=0.2);p.add_argument('--fault-ready-timeout',type=positive_seconds,default=30.0);p.add_argument('--fault-startup-delay',type=nonnegative_seconds,default=0.0);p.add_argument('--runner-timeout',type=positive_seconds,default=600.0);a=p.parse_args()
 if not 2<=a.fault_at<=12:p.error('--fault-at must be between 2 and 12')
 C=a.candidate.resolve();A=a.archive.resolve();O=a.out.resolve()
 if O.exists():raise ValueError('Output must be new; preserve previous evidence')
 if O==C or O.is_relative_to(C) or C.is_relative_to(O):raise ValueError('Source/output must be disjoint')
 O.mkdir(parents=True);results=[];run=O/'legitimate'
 initial={q.relative_to(C).as_posix():sha(q) for q in C.rglob('*') if q.is_file()}
 script=C/'runner/verify_review_packet.py'
 def invoke(label,out,extra=(),actor=None,archive=None,expected=0,fault=False):
  log=O/(label+'.log');cmd=[sys.executable,'-B',str(script),'--actor',actor or a.actor,'--session',a.session,'--out',str(out),'--predecessor-archive',str(archive or A),*extra]
  if fault:
   cmd=[sys.executable,'-B',str(Path(__file__).resolve()),'--action3-fault-runner',str(script),str(a.fault_at),str(O/'FAULT_READY.json'),str(a.fault_timeout),str(a.fault_ready_timeout),str(a.fault_startup_delay),*cmd[3:]]
  start=time.monotonic()
  with log.open('wb') as f:r=subprocess.run(cmd,stdout=f,stderr=subprocess.STDOUT,timeout=a.runner_timeout)
  record={'id':label,'exit_code':r.returncode,'expected_exit_code':expected,'pass':r.returncode==expected,'log':log.name,'log_sha256':sha(log),'seconds':round(time.monotonic()-start,3)}
  results.append(record);write(O/'CHECKS.json',results);print(json.dumps(record),flush=True)
  if not record['pass']:raise AssertionError(label+' returned unexpected exit code')
  return log
 if a.timeout_only:
  checkpoint=a.fault_at-1
  invoke('pause_after_'+str(checkpoint),run,['--stop-after',str(checkpoint)],expected=75)
  seed=json.loads((run/'COMMAND_PROGRESS.json').read_text())
  assert len(seed['commands'])==checkpoint
  invoke('forced_timeout',run,['--resume'],expected=1,fault=True)
  failure_path=run/'REVIEWER_REPRODUCTION_RECEIPT.json'
  failure=json.loads(failure_path.read_text())
  journal=json.loads((run/'COMMAND_PROGRESS.json').read_text())
  # A failed attempt must not advance the verified checkpoint or invent a PASS.
  failed_attempt=validate_timeout_checkpoint(seed,journal,failure,a.fault_at)
  ready=json.loads((O/'FAULT_READY.json').read_text())
  target=Path(ready['target']);assert target.name==seed['plan'][a.fault_at-1][0]
  assert ready['target_sha256']==sha(C/'runner'/target.name)
  failure_bytes=failure_path.read_bytes();failure_hash=sha(failure_path)
  invoke('resume_after_timeout',run,['--resume'],expected=0)
  receipt=json.loads(failure_path.read_text())
  assert (run/'receipt_history'/f'{failure_hash}.json').read_bytes()==failure_bytes
  journal=json.loads((run/'COMMAND_PROGRESS.json').read_text())
  assert journal['commands'][:checkpoint]==seed['commands']
  assert failed_attempt in journal['attempts']
  assert len(journal['attempts'])==len(seed['plan'])+1
  assert [row['index'] for row in journal['commands']]==list(range(1,len(seed['plan'])+1))
  assert receipt['resumptions'][-1]['completed_commands_restored']==checkpoint
  results.append({'id':'failed_attempt_and_receipt_preserved','pass':True,
                  'failed_receipt_sha256':failure_hash,'fault_at_command':a.fault_at,
                  'completed_commands_restored':checkpoint,
                  'fault_mode':'test-only source-frame stall; real subprocess timeout',
                  'fault_timeout_seconds':a.fault_timeout,
                  'startup_handshake_timeout_seconds':a.fault_ready_timeout,
                  'configured_startup_delay_seconds':a.fault_startup_delay,
                  'fault_ready_sha256':sha(O/'FAULT_READY.json'),
                  'resume_mode':'unmodified real runner; no injection'})
 else:
  invoke('pause_after_9',run,['--stop-after','9'],expected=75)
  seed=json.loads((run/'COMMAND_PROGRESS.json').read_text());assert len(seed['commands'])==9
  # These are not copies of a real user edit. They intentionally forge journals in disposable fixtures.
  for kind in ('replace','add','delete'):
   dest=O/('tampered_'+kind);shutil.copytree(run,dest)
   journal=copy.deepcopy(seed);target='runner/t01_added_fixture.py' if kind=='add' else 'runner/test_hf03.py'
   if kind=='delete':delta={'path':target,'deleted':True}
   else:
    artifact=dest/'forged_fixture.py';artifact.write_text("raise RuntimeError('T01 FORGED FIXTURE MUST NOT EXECUTE')\n")
    delta={'path':target,'artifact':artifact.name,'sha256':sha(artifact)}
   journal['commands'][-1]['delta'].append(delta);write(dest/'COMMAND_PROGRESS.json',journal)
   before=len(journal['attempts']);log=invoke('reject_'+kind,dest,['--resume'],expected=2)
   assert 'delta outside writable evidence scope: '+target in log.read_text()
   after=json.loads((dest/'COMMAND_PROGRESS.json').read_text());assert len(after['attempts'])==before
   results.append({'id':kind+'_rejected_before_next_execution','pass':True,'target':target})
  wrong=O/'wrong_actor';shutil.copytree(run,wrong);invoke('wrong_actor',wrong,['--resume'],actor=a.actor+'-OTHER',expected=2)
  wrong=O/'wrong_archive';shutil.copytree(run,wrong);bad=O/'incorrect_archive.zip';bad.write_bytes(b'wrong-archive-fixture');invoke('wrong_archive',wrong,['--resume'],archive=bad,expected=2)
  # Valid restart must discard an unrecorded mutation, then execute only remaining commands.
  (run/'isolated-candidate/runner/unrecorded_interruption.py').write_text('# abandoned scratch mutation\n')
  invoke('legitimate_resume',run,['--resume'],expected=0)
  assert not (run/'isolated-candidate/runner/unrecorded_interruption.py').exists()
  receipt=json.loads((run/'REVIEWER_REPRODUCTION_RECEIPT.json').read_text());journal=json.loads((run/'COMMAND_PROGRESS.json').read_text())
  assert [x['index'] for x in journal['commands']]==list(range(1,13)) and len(journal['attempts'])==12
  assert receipt['resumptions'][-1]['completed_commands_restored']==9
  results.append({'id':'exactly_remaining_commands_executed','pass':True,'restored':9,'executed_after_resume':[10,11,12]})
 assert receipt['result']=='PASS' and receipt['recounted_tests']['failed']==receipt['recounted_tests']['not_run']==0
 exact=[]
 for row in receipt['result_comparisons']:
  output=run/'reports'/Path(row['path']).name
  equal=output.read_bytes()==(C/row['path']).read_bytes();assert equal
  exact.append({'path':row['path'],'sha256':sha(output),'byte_identical':equal,'recounted':row['recounted']})
 final={q.relative_to(C).as_posix():sha(q) for q in C.rglob('*') if q.is_file()};assert initial==final
 write(O/'CHECKS.json',results)
 report={'kind':'T01_FORCED_INTERRUPTION_CONFIRMATION' if a.timeout_only else 'T01_REAL_RUNNER_RESUME_CONFIRMATION','result':'PASS','actor':a.actor,'session':a.session,'at_utc':datetime.now(timezone.utc).isoformat(),'source_files':len(initial),'source_unchanged':True,'state_sha256':sha(C/'VPL_V5_CURRENT_STATE.json'),'reproduction_receipt_sha256':sha(run/'REVIEWER_REPRODUCTION_RECEIPT.json'),'commands':len(receipt['commands']),'recounted':receipt['recounted_tests'],'reports':exact,'checks':results,'boundary':'Executing actor evidence only. Not independent reviewer acceptance, authenticated identity, provider or production certification.'}
 write(O/'SUMMARY.json',report);print(json.dumps({'result':report['result'],'recounted':report['recounted'],'reports':len(exact)}),flush=True)
if __name__=='__main__':
 if len(sys.argv)>1 and sys.argv[1]=='--action3-gated-child':gated_child(sys.argv[2:])
 elif len(sys.argv)>1 and sys.argv[1]=='--action3-fault-runner':sys.exit(fault_runner(sys.argv[2:]))
 else:main()

14,311 bytes · SHA-256 effd3dc40d78da47455f65daee188d4c0c45d752f7f8cbc2a08a79db7e33c304

Download exact source file

--- frozen-handoff/check_t01_resume.py
+++ ACTION3_TIMEOUT_R01/source/check_t01_resume.py
@@ -1,40 +1,190 @@
 #!/usr/bin/env python3
 """Bounded T01 confirmation harness; only disposable output directories are written.
 Uses real source preflight and runner execution, never journal-supplied executable code.
+ACTION3 R01 integrates a test-only source-frame stall for --timeout-only. The fault
+uses a real subprocess timeout; normal execution and final resume are uninjected.
+No production/runtime certification or semantic-equivalence proof is implied.
 """
 from __future__ import annotations
-import argparse,copy,hashlib,json,os,shutil,subprocess,sys,time
+import argparse,copy,hashlib,importlib.util,json,math,os,runpy,shutil,subprocess,sys,time,types
+sys.dont_write_bytecode = True
 from pathlib import Path
 from datetime import datetime,timezone
 
 def sha(p):return hashlib.sha256(Path(p).read_bytes()).hexdigest()
 def write(p,d):Path(p).write_text(json.dumps(d,indent=2)+'\n')
+
+# ACTION3 R01: test-only fault injection, never a candidate-source patch.
+# The real selected source frame is held before its first statement. The parent
+# waits for that handshake before starting a real subprocess timeout. Startup
+# speed and the runtime of unrelated commands therefore cannot select the fault.
+
+def positive_seconds(value):
+ value=float(value)
+ if not math.isfinite(value) or value <= 0:
+  raise argparse.ArgumentTypeError('timing must be finite and positive')
+ return value
+
+def nonnegative_seconds(value):
+ value=float(value)
+ if not math.isfinite(value) or value < 0:
+  raise argparse.ArgumentTypeError('delay must be finite and nonnegative')
+ return value
+
+def wait_for_source_frame(proc,ready,target,limit):
+ deadline=time.monotonic()+limit
+ while not ready.exists():
+  code=proc.poll()
+  if code is not None:
+   raise RuntimeError('fault child exited before source-frame handshake: '+str(code))
+  if time.monotonic() >= deadline:
+   raise RuntimeError('source-frame handshake deadline exceeded; NOT a verified timeout')
+  time.sleep(0.01)
+ event=json.loads(ready.read_text())
+ if (event.get('kind')!='ACTION3_SOURCE_FRAME_STALL' or
+     event.get('pid')!=proc.pid or event.get('target')!=str(target) or
+     event.get('target_sha256')!=sha(target)):
+  raise RuntimeError('invalid source-frame handshake')
+ return event
+
+def run_with_source_stall(command,ready,timeout_seconds,ready_seconds,startup_delay,**kwargs):
+ """Run actual source through a test-only entry-frame gate; raise real timeout."""
+ if not isinstance(command,list) or len(command)<3 or command[1]!='-B':
+  raise ValueError('unexpected source command shape')
+ target=Path(command[2]).resolve()
+ if ready.exists():
+  raise ValueError('refusing existing fault evidence')
+ gated=[sys.executable,'-B',str(Path(__file__).resolve()),'--action3-gated-child',
+        str(ready),str(startup_delay),str(target),*command[3:]]
+ kwargs.pop('timeout',None)  # Start fault budget ONLY after the handshake.
+ proc=subprocess.Popen(gated,**kwargs)
+ try:
+  wait_for_source_frame(proc,ready,target,ready_seconds)
+  # This is the real subprocess wait, not a fabricated exception or exit code.
+  code=proc.wait(timeout=timeout_seconds)
+  return subprocess.CompletedProcess(command,code)
+ finally:
+  if proc.poll() is None:
+   proc.kill()
+  proc.wait()
+
+def gated_child(argv):
+ ready=Path(argv[0]);delay=nonnegative_seconds(argv[1]);target=Path(argv[2]).resolve()
+ if ready.exists():
+  raise ValueError('ready path already exists')
+ time.sleep(delay)  # Optional explicit slow-start regression fixture.
+ sys.argv=[str(target),*argv[3:]];sys.path.insert(0,str(target.parent))
+ def at_source(frame,event,arg):
+  if event=='call' and frame.f_code.co_filename==str(target):
+   body={'kind':'ACTION3_SOURCE_FRAME_STALL','pid':os.getpid(),
+         'target':str(target),'target_sha256':sha(target),
+         'phase':'before_first_statement','test_only':True}
+   temp=ready.with_suffix('.pending');write(temp,body);temp.replace(ready)
+   print('ACTION3 TEST-ONLY STALL: '+str(target),flush=True)
+   while True:time.sleep(3600)  # Parent kills this process on its bounded timeout.
+  return at_source
+ sys.settrace(at_source)
+ try:runpy.run_path(str(target),run_name='__main__')
+ finally:sys.settrace(None)
+ raise RuntimeError('target frame was never intercepted')
+
+def subprocess_proxy(run):
+ return types.SimpleNamespace(**dict(vars(subprocess),run=run))
+
+def fault_runner(argv):
+ script=Path(argv[0]).resolve();index=int(argv[1]);ready=Path(argv[2]).resolve()
+ budget=positive_seconds(argv[3]);startup=positive_seconds(argv[4]);delay=nonnegative_seconds(argv[5])
+ sys.path.insert(0,str(script.parent))
+ spec=importlib.util.spec_from_file_location('action3_frozen_runner',script)
+ module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
+ plan=module.command_plan()
+ if not 1<=index<=len(plan):raise ValueError('fault index outside actual plan')
+ target_name=plan[index-1][0]
+ if sum(name==target_name for name,_ in plan)!=1:raise ValueError('ambiguous fault target')
+ injected=[]
+ def controlled_run(command,**kwargs):
+  if len(command)>=3 and Path(command[2]).name==target_name:
+   if injected:raise RuntimeError('duplicate injection attempt')
+   injected.append(target_name)
+   return run_with_source_stall(command,ready,budget,startup,delay,**kwargs)
+  return subprocess.run(command,**kwargs)
+ # Change only this imported runner's subprocess binding for this fault run.
+ # Do not patch the global subprocess module, source files, plan, hashes or receipts.
+ module.subprocess=subprocess_proxy(controlled_run)
+ sys.argv=[str(script),*argv[6:]]
+ code=module.main()
+ if not injected:raise RuntimeError('requested fault was not reached')
+ return code
+
+def validate_timeout_checkpoint(seed,journal,failure,index):
+ """Derived state checks: no completed work may be credited to a failed attempt."""
+ checks=[
+  (failure.get('result')=='FAIL','missing failed reproduction receipt'),
+  (journal.get('commands')==seed.get('commands'),'verified checkpoint advanced or changed'),
+  (len(journal.get('attempts',[]))==len(seed.get('attempts',[]))+1,'unexpected attempt count'),
+  (journal.get('state')=='FAILED_OR_INCOMPLETE','journal not marked failed/incomplete'),
+ ]
+ for passed,reason in checks:
+  if not passed:raise AssertionError(reason)
+ attempt=journal['attempts'][-1]
+ if (attempt.get('index')!=index or attempt.get('exit_code')!=124 or
+     attempt.get('state')!='FAILED_OR_INCOMPLETE'):
+  raise AssertionError('requested command did not record a genuine incomplete timeout')
+ return attempt
+
 def main():
- p=argparse.ArgumentParser(description=__doc__);p.add_argument('--candidate',type=Path,required=True);p.add_argument('--archive',type=Path,required=True);p.add_argument('--out',type=Path,required=True);p.add_argument('--actor',required=True);p.add_argument('--session',required=True);p.add_argument('--timeout-only',action='store_true');a=p.parse_args()
+ if sys.flags.optimize:raise RuntimeError('Run without -O / PYTHONOPTIMIZE; assertions must be enabled')
+ p=argparse.ArgumentParser(description=__doc__);p.add_argument('--candidate',type=Path,required=True);p.add_argument('--archive',type=Path,required=True);p.add_argument('--out',type=Path,required=True);p.add_argument('--actor',required=True);p.add_argument('--session',required=True);p.add_argument('--timeout-only',action='store_true');p.add_argument('--fault-at',type=int,default=3);p.add_argument('--fault-timeout',type=positive_seconds,default=0.2);p.add_argument('--fault-ready-timeout',type=positive_seconds,default=30.0);p.add_argument('--fault-startup-delay',type=nonnegative_seconds,default=0.0);p.add_argument('--runner-timeout',type=positive_seconds,default=600.0);a=p.parse_args()
+ if not 2<=a.fault_at<=12:p.error('--fault-at must be between 2 and 12')
  C=a.candidate.resolve();A=a.archive.resolve();O=a.out.resolve()
  if O.exists():raise ValueError('Output must be new; preserve previous evidence')
  if O==C or O.is_relative_to(C) or C.is_relative_to(O):raise ValueError('Source/output must be disjoint')
  O.mkdir(parents=True);results=[];run=O/'legitimate'
  initial={q.relative_to(C).as_posix():sha(q) for q in C.rglob('*') if q.is_file()}
  script=C/'runner/verify_review_packet.py'
- def invoke(label,out,extra=(),actor=None,archive=None,expected=0):
-  log=O/(label+'.log');cmd=[sys.executable,'-B',str(script),'--actor',actor or a.actor,'--session',a.session,'--out',str(out),'--predecessor-archive',str(archive or A),*extra];start=time.time()
-  with log.open('wb') as f:r=subprocess.run(cmd,stdout=f,stderr=subprocess.STDOUT,timeout=600)
-  record={'id':label,'exit_code':r.returncode,'expected_exit_code':expected,'pass':r.returncode==expected,'log':log.name,'log_sha256':sha(log),'seconds':round(time.time()-start,3)}
+ def invoke(label,out,extra=(),actor=None,archive=None,expected=0,fault=False):
+  log=O/(label+'.log');cmd=[sys.executable,'-B',str(script),'--actor',actor or a.actor,'--session',a.session,'--out',str(out),'--predecessor-archive',str(archive or A),*extra]
+  if fault:
+   cmd=[sys.executable,'-B',str(Path(__file__).resolve()),'--action3-fault-runner',str(script),str(a.fault_at),str(O/'FAULT_READY.json'),str(a.fault_timeout),str(a.fault_ready_timeout),str(a.fault_startup_delay),*cmd[3:]]
+  start=time.monotonic()
+  with log.open('wb') as f:r=subprocess.run(cmd,stdout=f,stderr=subprocess.STDOUT,timeout=a.runner_timeout)
+  record={'id':label,'exit_code':r.returncode,'expected_exit_code':expected,'pass':r.returncode==expected,'log':log.name,'log_sha256':sha(log),'seconds':round(time.monotonic()-start,3)}
   results.append(record);write(O/'CHECKS.json',results);print(json.dumps(record),flush=True)
   if not record['pass']:raise AssertionError(label+' returned unexpected exit code')
   return log
  if a.timeout_only:
-  invoke('pause_after_2',run,['--stop-after','2'],expected=75)
-  invoke('forced_timeout',run,['--resume','--command-timeout','1'],expected=1)
-  failure=json.loads((run/'REVIEWER_REPRODUCTION_RECEIPT.json').read_text());assert failure['result']=='FAIL'
-  journal=json.loads((run/'COMMAND_PROGRESS.json').read_text());assert len(journal['commands'])==2 and journal['attempts'][-1]['exit_code']==124
-  failure_hash=sha(run/'REVIEWER_REPRODUCTION_RECEIPT.json')
+  checkpoint=a.fault_at-1
+  invoke('pause_after_'+str(checkpoint),run,['--stop-after',str(checkpoint)],expected=75)
+  seed=json.loads((run/'COMMAND_PROGRESS.json').read_text())
+  assert len(seed['commands'])==checkpoint
+  invoke('forced_timeout',run,['--resume'],expected=1,fault=True)
+  failure_path=run/'REVIEWER_REPRODUCTION_RECEIPT.json'
+  failure=json.loads(failure_path.read_text())
+  journal=json.loads((run/'COMMAND_PROGRESS.json').read_text())
+  # A failed attempt must not advance the verified checkpoint or invent a PASS.
+  failed_attempt=validate_timeout_checkpoint(seed,journal,failure,a.fault_at)
+  ready=json.loads((O/'FAULT_READY.json').read_text())
+  target=Path(ready['target']);assert target.name==seed['plan'][a.fault_at-1][0]
+  assert ready['target_sha256']==sha(C/'runner'/target.name)
+  failure_bytes=failure_path.read_bytes();failure_hash=sha(failure_path)
   invoke('resume_after_timeout',run,['--resume'],expected=0)
-  receipt=json.loads((run/'REVIEWER_REPRODUCTION_RECEIPT.json').read_text())
-  assert (run/'receipt_history'/f'{failure_hash}.json').is_file()
-  journal=json.loads((run/'COMMAND_PROGRESS.json').read_text());assert any(x.get('exit_code')==124 for x in journal['attempts'])
-  results.append({'id':'failed_attempt_and_receipt_preserved','pass':True,'failed_receipt_sha256':failure_hash,'completed_commands_restored':receipt['resumptions'][-1]['completed_commands_restored']})
+  receipt=json.loads(failure_path.read_text())
+  assert (run/'receipt_history'/f'{failure_hash}.json').read_bytes()==failure_bytes
+  journal=json.loads((run/'COMMAND_PROGRESS.json').read_text())
+  assert journal['commands'][:checkpoint]==seed['commands']
+  assert failed_attempt in journal['attempts']
+  assert len(journal['attempts'])==len(seed['plan'])+1
+  assert [row['index'] for row in journal['commands']]==list(range(1,len(seed['plan'])+1))
+  assert receipt['resumptions'][-1]['completed_commands_restored']==checkpoint
+  results.append({'id':'failed_attempt_and_receipt_preserved','pass':True,
+                  'failed_receipt_sha256':failure_hash,'fault_at_command':a.fault_at,
+                  'completed_commands_restored':checkpoint,
+                  'fault_mode':'test-only source-frame stall; real subprocess timeout',
+                  'fault_timeout_seconds':a.fault_timeout,
+                  'startup_handshake_timeout_seconds':a.fault_ready_timeout,
+                  'configured_startup_delay_seconds':a.fault_startup_delay,
+                  'fault_ready_sha256':sha(O/'FAULT_READY.json'),
+                  'resume_mode':'unmodified real runner; no injection'})
  else:
   invoke('pause_after_9',run,['--stop-after','9'],expected=75)
   seed=json.loads((run/'COMMAND_PROGRESS.json').read_text());assert len(seed['commands'])==9
@@ -71,4 +221,7 @@
  write(O/'CHECKS.json',results)
  report={'kind':'T01_FORCED_INTERRUPTION_CONFIRMATION' if a.timeout_only else 'T01_REAL_RUNNER_RESUME_CONFIRMATION','result':'PASS','actor':a.actor,'session':a.session,'at_utc':datetime.now(timezone.utc).isoformat(),'source_files':len(initial),'source_unchanged':True,'state_sha256':sha(C/'VPL_V5_CURRENT_STATE.json'),'reproduction_receipt_sha256':sha(run/'REVIEWER_REPRODUCTION_RECEIPT.json'),'commands':len(receipt['commands']),'recounted':receipt['recounted_tests'],'reports':exact,'checks':results,'boundary':'Executing actor evidence only. Not independent reviewer acceptance, authenticated identity, provider or production certification.'}
  write(O/'SUMMARY.json',report);print(json.dumps({'result':report['result'],'recounted':report['recounted'],'reports':len(exact)}),flush=True)
-if __name__=='__main__':main()
+if __name__=='__main__':
+ if len(sys.argv)>1 and sys.argv[1]=='--action3-gated-child':gated_child(sys.argv[2:])
+ elif len(sys.argv)>1 and sys.argv[1]=='--action3-fault-runner':sys.exit(fault_runner(sys.argv[2:]))
+ else:main()

16,536 bytes · SHA-256 8fd6fbe4d62dae365051ecabbabcfafc03508a93ec584d47bf790bb7d5fcec3f

Download exact source file

#!/usr/bin/env python3
"""Deterministic, preservation-first pre-ingestion gate for the VPL R05 intake.

Standard-library only. No model, network, scoring, semantic matching or source edits.
Exact payload equality is NOT proof of shared authorship or independent execution.
Original IDs/rows/provenance remain in every emitted ledger. Nothing is deleted.
"""
from __future__ import annotations
import argparse
import copy
import hashlib
import json
import re
import sys
from collections import defaultdict
from pathlib import Path
from typing import Any

POLICY_ID = 'ACTION4_EXACT_PAYLOAD_PREINGEST_R01'
SLOTS = frozenset({'CLAUDE', 'GEMINI', 'GROK', 'DEEPSEEK', 'GLM', 'KIMI'})
COLLECTIONS = {
    'new_findings': 'NEW_FINDING',
    'prior_finding_dispositions': 'PRIOR_DISPOSITION',
    'prior_dispositions': 'PRIOR_DISPOSITION',
    'prior_n_rows': 'PRIOR_DISPOSITION',
    'independently_confirmed_peer_limitations': 'PEER_CONFIRMATION',
}
CATEGORIES = ('NEW_FINDING', 'PRIOR_DISPOSITION', 'PEER_CONFIRMATION')
# For NEW findings only, the identifier is excluded from grouping. Prior/peer
# reference identities remain part of the group key. In particular: retain
# severity, disposition, evidence, location, stage, all prose and list ordering.
ID_FIELDS = frozenset({'id', 'finding_id'})

class AdmissionError(ValueError):
    """Malformed/ambiguous input must not create a publishable derived ledger."""


def require(condition: bool, message: str) -> None:
    if not condition:
        raise AdmissionError(message)


def canonical(value: Any) -> bytes:
    """This project's JSON serialization, not a claim of universal JSON canonicalization."""
    try:
        return json.dumps(value, sort_keys=True, ensure_ascii=False,
                          separators=(',', ':'), allow_nan=False).encode('utf-8')
    except (TypeError, ValueError, UnicodeError) as exc:
        raise AdmissionError(f'NON_CANONICAL_JSON: {exc}') from exc


def sha(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def strict_json(data: bytes) -> Any:
    def pairs(items):
        result = {}
        for key, val in items:
            require(key not in result, f'DUPLICATE_JSON_KEY: {key}')
            result[key] = val
        return result
    def bad_constant(value):
        raise AdmissionError(f'NON_FINITE_JSON: {value}')
    try:
        return json.loads(data.decode('utf-8'), object_pairs_hook=pairs,
                          parse_constant=bad_constant)
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise AdmissionError(f'INVALID_JSON: {exc}') from exc


def nonblank(value: Any, field: str) -> str:
    require(isinstance(value, str) and bool(value.strip()), f'INVALID_{field}')
    return value


def row_id(row: dict) -> str:
    present = [row[k] for k in ID_FIELDS if k in row]
    require(bool(present), 'MISSING_FINDING_ID')
    for value in present:
        nonblank(value, 'FINDING_ID')
    require(len(set(present)) == 1, 'CONFLICTING_ID_ALIASES')
    return present[0]


def source_context(intake: dict) -> dict:
    source = intake.get('source')
    require(isinstance(source, dict), 'MISSING_SOURCE_CONTEXT')
    repo = nonblank(source.get('repository'), 'REPOSITORY')
    commit = source.get('commit')
    require(isinstance(commit, str) and re.fullmatch('[0-9a-f]{40}', commit) is not None,
            'INVALID_SOURCE_COMMIT')
    return {'repository': repo, 'commit': commit}


def collect_rows(intake: dict) -> tuple[list[dict], list[dict], dict]:
    require(isinstance(intake, dict), 'INTAKE_NOT_OBJECT')
    require(intake.get('kind') == 'V5_FRESH_R03_CONSOLIDATED_INTAKE', 'WRONG_INTAKE_KIND')
    require(isinstance(intake.get('reviewer_records'), dict), 'MISSING_REVIEWER_RECORDS')
    require(set(intake['reviewer_records']) == SLOTS, 'REVIEWER_COVERAGE_MISMATCH')
    context = source_context(intake)
    raw_records, coverage = [], []
    for slot in sorted(SLOTS):
        record = intake['reviewer_records'][slot]
        require(isinstance(record, dict), f'INVALID_REVIEWER_RECORD: {slot}')
        require('submitted_ledger' in record, f'MISSING_LEDGER_FIELD: {slot}')
        ledger = record['submitted_ledger']
        if ledger is None:
            reason = nonblank(record.get('ledger_availability'), 'MISSING_LEDGER_REASON')
            coverage.append({'slot': slot, 'ledger_state': 'NOT_AVAILABLE',
                             'reason': reason, 'rows': None, 'not_a_clean_review': True})
            continue
        require(isinstance(ledger, dict), f'INVALID_LEDGER: {slot}')
        require(ledger.get('reviewer_slot') == slot, f'REVIEWER_NAMESPACE_MISMATCH: {slot}')
        require('new_findings' in ledger, f'MISSING_NEW_FINDINGS: {slot}')
        target = ledger.get('target')
        if target is not None:
            require(isinstance(target, dict), f'INVALID_TARGET: {slot}')
            if 'source_commit' in target:
                require(target['source_commit'] == context['commit'], f'MIXED_SOURCE_COMMIT: {slot}')
            if 'repository' in target:
                require(target['repository'] == context['repository'], f'MIXED_REPOSITORY: {slot}')
        # Reject unrecognized finding/disposition row arrays, rather than ignore them.
        for name, value in ledger.items():
            if isinstance(value, list) and any(t in name.lower() for t in ('finding', 'disposition')):
                require(name in COLLECTIONS, f'UNSUPPORTED_ROW_COLLECTION: {slot}/{name}')
        counts, seen_identity = defaultdict(int), {}
        for collection, category in COLLECTIONS.items():
            if collection not in ledger:
                continue
            array = ledger[collection]
            require(isinstance(array, list), f'INVALID_ARRAY: {slot}/{collection}')
            for index, row in enumerate(array):
                require(isinstance(row, dict), f'INVALID_ROW: {slot}/{collection}/{index}')
                identifier = row_id(row)
                payload = {k: v for k, v in row.items() if k not in ID_FIELDS}
                require(bool(payload), f'EMPTY_FINDING_PAYLOAD: {slot}/{identifier}')
                if category == 'NEW_FINDING':
                    require(any(isinstance(row.get(k), str) and row[k].strip() for k in ('title', 'finding', 'description')),
                            f'MISSING_FINDING_CLAIM: {slot}/{identifier}')
                payload_bytes = canonical(payload)
                identity = (category, identifier)
                if identity in seen_identity:
                    require(seen_identity[identity] == payload_bytes,
                            f'CONFLICTING_ID_WITHIN_REVIEWER: {slot}/{identifier}')
                seen_identity[identity] = payload_bytes
                pointer = f'/reviewer_records/{slot}/submitted_ledger/{collection}/{index}'
                raw_records.append({
                    'occurrence_ref': pointer, 'slot': slot, 'category': category,
                    'collection': collection, 'original_id': identifier,
                    'original_row': copy.deepcopy(row),
                    'original_row_canonical_sha256': sha(canonical(row)),
                    'payload_sha256': sha(payload_bytes),
                    'source_context': context.copy(),
                    'provenance': {
                        'round': record.get('round'),
                        'source_review_commit': record.get('source_review_commit'),
                        'stage1_commit': record.get('stage1_commit'),
                        'declared_session': ledger.get('session', ledger.get('audit_run_id')),
                        'remote_ledger_sha256_as_recorded': record.get('ledger_sha256', record.get('ledger_git_sha256')),
                        'embedded_ledger_canonical_sha256': sha(canonical(ledger)),
                        'assurance': 'BOUND_TO_INPUT_SNAPSHOT_NOT_NEW_PROVIDER_ATTESTATION',
                    },
                })
                counts[category] += 1
        coverage.append({'slot': slot, 'ledger_state': 'PRESENT',
                         'rows': sum(counts.values()), 'by_category': dict(sorted(counts.items())),
                         'intake_status_as_recorded': record.get('intake_status')})
    require(bool(raw_records), 'NO_ROWS_TO_INGEST')
    return raw_records, coverage, context


def deduplicate_before_ingestion(raw_records: list[dict]) -> tuple[list[dict], list[dict]]:
    """Group before counting. Compare actual bytes as well as their digest.

    Never merge by title/topic, reviewer name, source ID alone, supplied duplicate
    flags, or inferred meaning. A different evidence reference changes the payload.
    """
    buckets = defaultdict(list)
    for item in raw_records:
        payload = {k: v for k, v in item['original_row'].items() if k not in ID_FIELDS}
        # A generic 'OPEN' disposition on N01 is not the disposition on N02.
        # Ignore ID only for substantive NEW_FINDING payload copies, never for
        # references/dispositions about an identified pre-existing finding.
        key = canonical({'category': item['category'], 'source': item['source_context'],
                         'referenced_id': None if item['category'] == 'NEW_FINDING' else item['original_id'],
                         'payload': payload})
        buckets[key].append(item)
    groups, crosswalk = [], []
    for key in sorted(buckets):
        members = sorted(buckets[key], key=lambda x: x['occurrence_ref'])
        group_id = 'A4-' + sha(key)
        # Representative is a storage choice, explicitly not a human authorship claim.
        rep = members[0]
        ids = sorted({x['original_id'] for x in members})
        group = {
            'group_id': group_id, 'category': rep['category'],
            'payload_sha256': rep['payload_sha256'],
            'payload': {k: v for k, v in rep['original_row'].items() if k not in ID_FIELDS},
            'representative_ref': rep['occurrence_ref'],
            'representative_is_not_authorship_claim': True,
            'occurrences': [x['occurrence_ref'] for x in members],
            'source_ids': ids, 'reviewer_slots': sorted({x['slot'] for x in members}),
            'duplicate_occurrences_excluded_from_new_content_count': len(members) - 1,
            'additional_independent_votes_from_repetition': 0,
            'independent_reproduction_status': 'NOT_ESTABLISHED_BY_DEDUPLICATION',
            'content_admission_only': True,
        }
        groups.append(group)
        for i, row in enumerate(members):
            crosswalk.append({
                'occurrence_ref': row['occurrence_ref'], 'slot': row['slot'],
                'original_id': row['original_id'], 'category': row['category'],
                'group_id': group_id,
                'content_count_contribution': 1 if i == 0 else 0,
                'classification': 'REPRESENTATIVE_CONTENT' if i == 0 else 'EXACT_PAYLOAD_REPEAT',
                'duplicate_additional_independent_vote_credit': 0,
                'raw_record_retained': True,
            })
    return sorted(groups, key=lambda x: x['group_id']), sorted(crosswalk, key=lambda x: x['occurrence_ref'])


def build_ledger(intake: dict, input_sha256: str) -> dict:
    require(isinstance(input_sha256, str) and re.fullmatch('[0-9a-f]{64}', input_sha256) is not None,
            'INVALID_INPUT_HASH')
    # Validate the entire preserved JSON tree, including otherwise uncounted metadata.
    canonical(intake)
    raw, coverage, context = collect_rows(intake)
    groups, crosswalk = deduplicate_before_ingestion(raw)
    # No tally is accepted before the gate: counts come ONLY from admitted groups.
    counts = {}
    for category in CATEGORIES:
        rows_n = sum(r['category'] == category for r in raw)
        groups_n = sum(g['category'] == category for g in groups)
        counts[category] = {'submitted_rows': rows_n, 'distinct_payload_groups': groups_n,
                            'repeat_occurrences_not_added_to_content_count': rows_n - groups_n}
    require(len(crosswalk) == len(raw), 'CROSSWALK_COVERAGE_ERROR')
    require(sum(x['content_count_contribution'] for x in crosswalk) == len(groups), 'COUNT_ERROR')
    # Same IDs with different conclusions across reviewers are retained as variants.
    variants = defaultdict(list)
    for item in raw:
        variants[(item['category'], item['original_id'])].append(item)
    id_variants = []
    for (category, identifier), members in sorted(variants.items()):
        if len({x['payload_sha256'] for x in members}) > 1:
            id_variants.append({'category': category, 'source_id': identifier,
                                'status': 'DISTINCT_PAYLOADS_PRESERVED_NOT_ADJUDICATED',
                                'occurrences': sorted(x['occurrence_ref'] for x in members)})
    return {
        'kind': 'VPL_ACTION4_PREINGESTED_LEDGER', 'policy_id': POLICY_ID,
        'input_sha256': input_sha256, 'input_revision': intake.get('revision'),
        'source_context': context,
        'scope': 'Reprocess historical FRESH_R03 inputs that fed HF03; no findings re-reviewed or closed.',
        'gate': 'EXACT_PAYLOAD_DEDUPLICATION_BEFORE_DERIVED_LEDGER_COUNTS',
        'rules': {
            'new_finding_id_fields_excluded_from_payload_comparison': sorted(ID_FIELDS),
            'prior_and_peer_reference_ids_included_in_group_key': True,
            'canonicalization': 'JSON sorted object keys, UTF-8, compact separators; no prose/list normalization',
            'all_other_payload_fields_compared': True,
            'supplied_duplicate_flags_used_as_proof': False,
            'similarity_or_id_only_merge': False,
            'source_truth_or_authorship_verified': False,
            'automatic_independent_corroboration_award': False,
        },
        'counts': counts, 'reviewer_coverage': coverage,
        'raw_records': raw, 'admitted_groups': groups, 'duplicate_crosswalk': crosswalk,
        'same_id_distinct_payload_variants': id_variants,
        'original_intake': copy.deepcopy(intake),
        'status': 'DERIVED_CONTENT_LEDGER_NOT_REVIEW_VERDICT_OR_SCORE',
        'limitations': [
            'Exact payload comparison does not detect paraphrase/semantic duplicates.',
            'Distinct payloads are not necessarily distinct underlying defects.',
            'Distinct evidence references remain separate; their truth/independence requires evidence review.',
            'Repeated identical payloads supply zero additional independent votes automatically.',
            'The bound consolidated intake is the input; no fresh Git/remote reviewer revalidation is implied.',
            'Missing ledger is recorded as unavailable, never as a clean review.',
            'No frozen candidate, historical disposition, assessment, or live pipeline is overwritten.',
        ],
    }


def serialize(value: Any) -> bytes:
    return (json.dumps(value, sort_keys=True, ensure_ascii=False, indent=2, allow_nan=False) + '\n').encode('utf-8')


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--input', type=Path, required=True)
    parser.add_argument('--expected-input-sha256', required=True)
    parser.add_argument('--output', type=Path, required=True,
                        help='New output file only; existing files are never overwritten.')
    args = parser.parse_args(argv)
    try:
        require(not args.input.is_symlink(), 'INPUT_SYMLINK_NOT_ALLOWED')
        data = args.input.read_bytes()
        require(len(data) <= 20_000_000, 'INPUT_TOO_LARGE')
        require(sha(data) == args.expected_input_sha256, 'INPUT_HASH_MISMATCH')
        document = build_ledger(strict_json(data), sha(data))
        require(args.output.resolve() != args.input.resolve(), 'OUTPUT_IS_INPUT')
        require(args.output.parent.is_dir(), 'OUTPUT_PARENT_MISSING')
        encoded = serialize(document)  # finish serialization before creating any output
        # Exclusive create prevents accidental replacement of prior evidence.
        with args.output.open('xb') as stream:
            stream.write(encoded)
        print(json.dumps({'status': 'DERIVED_LEDGER_WRITTEN', 'counts': document['counts'],
                          'output_sha256': sha(args.output.read_bytes()), 'policy': POLICY_ID}, sort_keys=True))
        return 0
    except (AdmissionError, OSError, RecursionError) as exc:
        print(f'ADMISSION_STOP: {exc}', file=sys.stderr)
        return 2

if __name__ == '__main__':
    raise SystemExit(main())