verify cycle.py

#!/usr/bin/env python3
"""Portable, independent replay of C1483--C1582.

Uses only files below this script's directory. Does not import research.py or
runner.py, edit source models, or amend the ordered journal. Decimal reversal
uses integer divmod on original durations. --require-100 closes final checks;
the default accepts an ordered completed prefix of the cycle.
"""
from __future__ import annotations

import argparse
from datetime import datetime
from fractions import Fraction
import hashlib
import json
from pathlib import Path
import subprocess
import sys

BASE=Path(__file__).resolve().parent
CHECKS=[]
CHECKED=set()
NOT_PRESENT=[]

def encode(value):
    if isinstance(value,Fraction):
        return value.numerator if value.denominator==1 else str(value)
    if isinstance(value,dict): return {str(k):encode(v) for k,v in value.items()}
    if isinstance(value,(list,tuple)): return [encode(v) for v in value]
    return value

def check(label,actual,expected=True):
    actual,expected=encode(actual),encode(expected)
    CHECKS.append({'name':label,'actual':actual,'expected':expected,'pass':actual==expected})

def read_json(path): return json.loads(path.read_text(encoding='utf-8'))
def sha(payload): return hashlib.sha256(payload).hexdigest()
def record_hash(record):
    core={k:v for k,v in record.items() if k!='sha256'}
    return sha(json.dumps(core,sort_keys=True,separators=(',',':'),ensure_ascii=False).encode())

def reverse_original(n):
    assert type(n) is int and n>0
    scale=1
    while n % 10 == 0:
        n//=10
        scale*=10
    result=0
    while n:
        n,digit=divmod(n,10)
        result=result*10+digit
    return result*scale

def rebuild(anchor,original_legs):
    out=[anchor]
    for duration in original_legs:
        out.append(out[-1]+reverse_original(duration))
    return out

def verify_model(name,expected,subset=False,optional=True):
    filename=name+'.json'
    path=BASE/'model'/filename
    if not path.exists():
        NOT_PRESENT.append(filename)
        if not optional: check('required model exists: '+filename,False)
        return
    actual=read_json(path)
    if subset: actual={key:actual.get(key) for key in expected}
    check('model/'+name,actual,expected)
    CHECKED.add(filename)

def verify_sources():
    binding_path=BASE/'model/source_bindings.json'
    bindings=read_json(binding_path)
    CHECKED.add('source_bindings.json')
    for name,binding in bindings.items():
        target=(BASE/'evidence'/name).resolve()
        check('source remains inside evidence: '+name,target.is_relative_to((BASE/'evidence').resolve()))
        check('source exists: '+name,target.is_file())
        if target.is_file():
            data=target.read_bytes()
            check('source bound bytes and SHA: '+name,[len(data),sha(data)],
                  [binding['bytes'],binding['sha256']])
    return bindings

def verify_journal(require100):
    parent=read_json(BASE/'evidence/C1482_parent_record.json')
    journal_bytes=(BASE/'journal.json').read_bytes()
    journal=json.loads(journal_bytes)
    check('parent is completed C1482',parent.get('step'),1482)
    check('parent canonical record hash',record_hash(parent),parent.get('sha256'))
    check('journal is ordered list',isinstance(journal,list))
    check('journal count bounds',0<len(journal)<=100)
    if require100:
        check('one hundred completed actions',len(journal),100)
        check('no pending action at close',not (BASE/'pending.json').exists())
    previous=parent['sha256']
    prior_closed=None
    required={'step','title','question','sources','opened_utc','predecessor_sha256',
              'results','finding','reassessment','checks','closed_utc','sha256'}
    for expected_step,record in enumerate(journal,1483):
        prefix='C'+str(expected_step)
        check(prefix+' complete fields',required.issubset(record))
        check(prefix+' sequence',record.get('step'),expected_step)
        check(prefix+' predecessor',record.get('predecessor_sha256'),previous)
        check(prefix+' canonical record hash',record_hash(record),record.get('sha256'))
        check(prefix+' explicit reassessment',bool(str(record.get('reassessment','')).strip()))
        declared=record.get('checks',{})
        check(prefix+' declared check results',all(v is True for v in declared.values()))
        opened=datetime.fromisoformat(record['opened_utc'])
        closed=datetime.fromisoformat(record['closed_utc'])
        check(prefix+' closure after opening',closed>=opened)
        if prior_closed is not None: check(prefix+' order in time',opened>=prior_closed)
        previous,prior_closed=record['sha256'],closed
        result=record.get('results',{})
        if isinstance(result,dict) and 'review_file' in result and 'review_sha256' in result:
            review=(BASE/result['review_file']).resolve()
            check(prefix+' review is portable',review.is_relative_to(BASE))
            check(prefix+' review exists',review.is_file())
            if review.is_file(): check(prefix+' review content hash',sha(review.read_bytes()),result['review_sha256'])
        if isinstance(result,dict) and 'strategy_sha256' in result:
            check(prefix+' bound Strategy edition',sha((BASE/'evidence/Strategy.md').read_bytes()),result['strategy_sha256'])
    if len(journal)>=50:
        check('C1532 Strategy checkpoint present',journal[49]['step'],1532)
        check('C1532 Strategy review recorded','review_sha256' in journal[49].get('results',{}))
    if require100:
        check('C1582 final step',journal[-1]['step'],1582)
        check('C1582 Strategy review recorded','review_sha256' in journal[-1].get('results',{}))
    return {'actions':len(journal),'first':journal[0]['step'],'last':journal[-1]['step'],
            'parent_record_sha256':parent['sha256'],'last_record_sha256':previous,
            'journal_file_sha256':sha(journal_bytes)}

def verify_cumulative(require100):
    # Declared source inputs independently transcribed from the frozen material.
    nodes={
        'MT':{'C':14006,'F':4836,'S':5436,'N':6386,'A':1406},
        'LXX':{'C':14896,'F':5746,'S':6346,'N':7296,'A':1406},
        'SP':{'C':13396,'F':4716,'S':5316,'N':6266,'A':1406},
    }
    verify_model('cumulative_source_nodes',nodes,optional=not require100)
    names=['Adam','Seth','Enosh','Kenan','Mahalalel','Jared','Enoch','Methuselah','Lamech']
    actual={'MT':[930,912,905,910,895,962,365,969,777],
            'LXX':[930,912,905,910,895,962,365,969,753],
            'SP':[930,912,905,910,895,847,365,720,653]}
    rounded={t:[5*((n+2)//5) for n in rows] for t,rows in actual.items()}
    sums={t:sum(rows) for t,rows in rounded.items()}
    residuals={t:[r-a for r,a in zip(rounded[t],actual[t])] for t in actual}
    verify_model('prefix_source_rows',{'names':names,'actual':actual,'rounded':rounded,
                                     'sums':sums,'residuals':residuals},optional=not require100)
    spec={'F':['F'],'S':['S'],'N':['N'],'FS':['F','S'],'FN':['F','N'],
          'SN':['S','N'],'FSN':['F','S','N']}
    verify_model('partition_spec',{'order':['A','F','S','N','C'],'partitions':spec,
                                  'head_and_anchor':'alwaysretained',
                                  'reversal':'eachoriginallegonce',
                                  'empty_selection':'notpartofthisnamedspineextension'},optional=not require100)
    partitions={}
    for tradition,v in nodes.items():
        check(tradition+' retained serial Noah and Shem lives',[v['N']-v['S'],v['S']-v['F']],[950,600])
        check(tradition+' rounded prefix agrees with selected head',v['C']-v['N'],sums[tradition]-(5 if tradition=='SP' else 0))
        paths={}
        for name,interior in spec.items():
            source=[v['A']]+[v[k] for k in interior]+[v['C']]
            legs=[b-a for a,b in zip(source,source[1:])]
            inv=[reverse_original(n) for n in legs]
            path1406=rebuild(1406,legs)
            path6=rebuild(6,[1400]+legs)
            paths[name]={'source_nodes':source,'source_legs':legs,'inverse_legs':inv,
                         'path1406':path1406,'path6':path6,'total':sum(inv)}
            check(tradition+'/'+name+' source total',[sum(legs),source[-1]-source[0]],[v['C']-1406]*2)
            check(tradition+'/'+name+' common stage translation',
                  [b-a for a,b in zip(path1406,path6[1:])],[2700]*len(path1406))
        verify_model('partitions_'+tradition,paths,optional=not require100)
        partitions[tradition]=paths
    lower={}
    for t,v in nodes.items():
        a=v['F']-1406
        lower[t]={'a':a,'merged_original':a+1550,'merged_inverse':reverse_original(a+1550),
                  'split_inverse':reverse_original(a)+reverse_original(1550),
                  'defect':reverse_original(a+1550)-reverse_original(a)-reverse_original(1550)}
        check(t+' named Noah lower merge',[lower[t]['defect'],partitions[t]['N']['total']-partitions[t]['FN']['total']],[0,0])
    verify_model('lower_1550_law',lower,optional=not require100)
    for t,p in partitions.items():
        check(t+' common explicit lifespan split',p['FN']['total']-p['FSN']['total'],4320)
    check('LXX refinement/completion join',partitions['LXX']['FSN']['path1406'][-1],partitions['LXX']['F']['path6'][-1])
    check('LXX joined coordinate',partitions['LXX']['FSN']['path1406'][-1],13636)
    check('SP source to completed refined head',nodes['SP']['C']-partitions['SP']['FSN']['path6'][-1],3600)
    check('MT inherited secondary Flood/Noah head',partitions['MT']['FN']['path1406'][-1],13016)
    check('LXX prefix reversal register contrast',reverse_original(7600)-reverse_original(7620),4030)
    split_loss=reverse_original(1550)-reverse_original(600)-reverse_original(950)
    verify_model('lifespan_split_law',{'original_combined':600+950,'original_separate':[600,950],
                 'inverse_combined':reverse_original(1550),
                 'inverse_separate':[reverse_original(600),reverse_original(950)],
                 'loss':split_loss},optional=not require100)
    upper={}
    refinement={}
    scope={}
    for t,v in nodes.items():
        prefix=v['C']-v['N']
        primary_upper=reverse_original(v['C']-v['F'])
        grouped_upper=reverse_original(1550)+reverse_original(prefix)
        gain=grouped_upper-primary_upper
        upper[t]={'prefix':prefix,'primary_upper':primary_upper,
                  'Noah_grouped_upper':grouped_upper,'gain':gain}
        refinement[t]={'upper_gain':gain,'split_loss':split_loss,
                       'predicted_change':gain-split_loss,
                       'observed_change':partitions[t]['FSN']['total']-partitions[t]['F']['total']}
        held=partitions[t]['FSN']['path1406'][-1]
        completed=partitions[t]['F']['path6'][-1]
        scope[t]={'refined_held':held,'primary_completed':completed,'signed_difference':held-completed}
    verify_model('upper_1550_law',upper,optional=not require100)
    verify_model('refinement_law',refinement,optional=not require100)
    verify_model('refinement_scope',scope,optional=not require100)
    lxx=partitions['LXX']
    verify_model('LXX_completion_join',{'refined_held_path':lxx['FSN']['path1406'],
                 'primary_completed_path':lxx['F']['path6'],
                 'refinement_gain':lxx['FSN']['total']-lxx['F']['total'],
                 'stage_gain':reverse_original(1400)-1400,
                 'common_head':lxx['FSN']['path1406'][-1]},optional=not require100)
    sp=partitions['SP']['FSN']
    source_branch=sp['source_nodes'][1:]
    inverse_branch=sp['path1406'][1:]
    verify_model('SP_refinement_scope',{'source_branch':source_branch,'inverse_branch':inverse_branch,
                 'node_shifts':[b-a for a,b in zip(source_branch,inverse_branch)],
                 'new_local_contractions':[950-reverse_original(950),7130-reverse_original(7130)]},
                 optional=not require100)
    def defect(u,v):
        return reverse_original(u+v)-reverse_original(u)-reverse_original(v)
    grammar={}
    for t,v in nodes.items():
        a,c=v['F']-1406,v['C']-v['N']
        total=sum(reverse_original(n) for n in (a,600,950,c))
        d_a600=defect(a,600)
        d_pair=defect(600,950)
        d_950c=defect(950,c)
        d_a1550=defect(a,1550)
        d_1550c=defect(1550,c)
        d_a600_950=defect(a+600,950)
        grammar[t]={'full_total':total,'D_a_600':d_a600,'D_600_950':d_pair,
                    'D_950_c':d_950c,'D_a_1550':d_a1550,'D_1550_c':d_1550c,
                    'D_a600_950':d_a600_950}
        generated={'F':total+d_pair+d_1550c,
                   'S':total+d_a600+d_950c,
                   'N':total+d_pair+d_a1550,
                   'FS':total+d_950c,
                   'FN':total+d_pair,
                   'SN':total+d_a600,
                   'FSN':total}
        check(t+' all seven totals regenerated from merging defects',generated,
              {name:p['total'] for name,p in partitions[t].items()})
        check(t+' lower merge cocycle',d_a600+d_a600_950,d_pair+d_a1550)
        check(t+' upper merge cocycle',d_pair+d_1550c,d_950c+defect(600,950+c))
        check(t+' cocycle alternative Noah total',total+d_a600+d_a600_950,partitions[t]['N']['total'])
    verify_model('merge_defect_grammar',grammar,optional=not require100)
    return nodes,partitions

def verify_4176_source_role():
    text29=(BASE/'evidence/File29.md').read_text()
    text17=(BASE/'evidence/File17.md').read_text()
    check('File29 appoints4176 Year6+60 state','`4176 BC` Year-6 +60 / Real state' in text29)
    check('File29 distinguishes4116 Year6 and4114 endpoint',
          '`4116 BC` Year-6 state' in text29 and '`4114 BC` endpoint state' in text29)
    check('File17 appoints4176 MT+275 Year6 state',
          '`4176 BC`, Year 6, MT `+275`' in text17)
    check('File17 names3901 Year6 variant base','4176n BC → 3901n BC = 275' in text17)
    check('Year6 variant descriptions agree',[4116+60,3901+215+60,3901+275],[4176]*3)
    check('Year6 source pair is translated SP pair',[2736+1440,6+1440],[4176,1446])
    check('Year6 source interval retains2730',4176-1446,2736-6)

def verify_babylonian():
    script=BASE/'agents/verify_babylonian.py'
    run=subprocess.run([sys.executable,str(script)],cwd=BASE,capture_output=True,text=True)
    check('Babylonian verifier exits successfully',run.returncode,0)
    result_path=BASE/'agents/babylonian_root_verification.json'
    if not result_path.is_file():
        check('Babylonian verification result exists',False)
        return {}
    child=read_json(result_path)
    check('Babylonian arithmetic status',child.get('status'),'PASS')
    for item in child.get('checks',[]):
        CHECKS.append({**item,'name':'Babylonian/'+item['name']})
    CHECKED.update(child.get('models_checked',[]))
    return {k:child.get(k) for k in ('status','grouped_check_count','models_checked','additional_model_files_not_checked')}

def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--require-100',action='store_true',help='Require C1483 through C1582 and no pending action')
    parser.add_argument('--output',type=Path,default=BASE/'verification.json')
    args=parser.parse_args()
    journal_info={}
    child={}
    try:
        verify_sources()
        journal_info=verify_journal(args.require_100)
        nodes,partitions=verify_cumulative(args.require_100)
        verify_4176_source_role()
        child=verify_babylonian()
    except Exception as error:
        check('verification execution',type(error).__name__+': '+str(error),'successful replay')
    # These models record research interpretation, not additional numeric claims.
    narrative={'babylonian_kernel.json','claim_ledger_first50.json',
               'claim_ledger_second50.json','final_kernel.json'}
    present={p.name for p in (BASE/'model').glob('*.json')}
    for name in sorted(present & narrative):
        value=read_json(BASE/'model'/name)
        check('narrative model is nonempty object: '+name,isinstance(value,dict) and bool(value))
    unverified=sorted(present-CHECKED-narrative)
    check('all arithmetic model categories covered',unverified,[])
    failed=[r for r in CHECKS if not r['pass']]
    result={'status':'PASS' if not failed else 'FAIL','require_100':args.require_100,
            'journal':journal_info,'grouped_check_count':len(CHECKS),'failed_check_count':len(failed),
            'models_checked':sorted(CHECKED),'narrative_models':sorted(present & narrative),
            'unverified_models':unverified,'not_yet_present':NOT_PRESENT,
            'Babylonian_verifier':child,
            'method':'Portable source SHA bindings, canonical linked-journal hashes, independent integer divmod reversal and exact Fraction arithmetic. Counts are verification groups, not discoveries.',
            'checks':CHECKS}
    args.output.parent.mkdir(parents=True,exist_ok=True)
    args.output.write_text(json.dumps(result,indent=2,ensure_ascii=False)+'\n')
    print(json.dumps({k:result[k] for k in ('status','require_100','journal','grouped_check_count','failed_check_count','unverified_models')}))
    if failed:
        for failure in failed: print(json.dumps(failure,ensure_ascii=False))
    return 0 if not failed else 1

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