verify.py

#!/usr/bin/env python3
"""Read-only replay of C532--C631, or a completed prefix via --through N.

The journal writer is never imported. Standard output is the JSON report;
--output may additionally save that report. Replay itself cannot write files.
The supplied packet is trusted research code, not an arbitrary-code sandbox.
"""
from pathlib import Path
from fractions import Fraction
from datetime import datetime
import argparse
import ast
import contextlib
import copy
import hashlib
import io
import json
import os
import runpy
import sys
import types

ROOT = Path(__file__).resolve().parent.parent
FIRST, LAST = 532, 631
PREDECESSOR_HASH = '646c8bb822355b60f1110fbb059453a1f0ebfaf7294c715f338ca64cf2bf5216'
PRIMARY_HASH = 'a5ea84562101158b60d0cf296765d6eff38e7a2abda4e74ad1b353dfd13b9530'
REPLAY_ACTIVE = False


class VerificationError(RuntimeError):
    pass


def require(condition, message):
    if not condition:
        raise VerificationError(message)


def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def record_digest(row):
    return hashlib.sha256(json.dumps({k: v for k, v in row.items() if k != 'sha256'}, sort_keys=True).encode()).hexdigest()


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


class InverseValue(int):
    """Track values descended from inv(), including ordinary integer arithmetic."""
    @staticmethod
    def wrap(value):
        return InverseValue(value) if isinstance(value, int) else value

    def __add__(self, other): return self.wrap(int.__add__(self, other))
    def __radd__(self, other): return self.wrap(int.__add__(self, other))
    def __sub__(self, other): return self.wrap(int.__sub__(self, other))
    def __rsub__(self, other): return self.wrap(int.__rsub__(self, other))
    def __mul__(self, other): return self.wrap(int.__mul__(self, other))
    def __rmul__(self, other): return self.wrap(int.__mul__(self, other))
    def __floordiv__(self, other): return self.wrap(int.__floordiv__(self, other))
    def __rfloordiv__(self, other): return self.wrap(int.__rfloordiv__(self, other))
    def __mod__(self, other): return self.wrap(int.__mod__(self, other))
    def __rmod__(self, other): return self.wrap(int.__rmod__(self, other))
    def __pow__(self, other, modulo=None): return self.wrap(int.__pow__(self, other, modulo))
    def __neg__(self): return InverseValue(int.__neg__(self))
    def __pos__(self): return InverseValue(self)
    def __abs__(self): return InverseValue(int.__abs__(self))


def audit_read_only(event, args):
    if not REPLAY_ACTIVE:
        return
    if event == 'open':
        mode = args[1] if len(args) > 1 else None
        flags = args[2] if len(args) > 2 else 0
        write_flags = os.O_WRONLY | os.O_RDWR | os.O_APPEND | os.O_CREAT | os.O_TRUNC
        require(not (isinstance(mode, str) and any(c in mode for c in 'wax+')), 'Replay attempted a write-capable open')
        require(not (isinstance(flags, int) and flags & write_flags), 'Replay attempted a write-capable file descriptor')
    forbidden = ('subprocess.', 'socket.', 'ctypes.', 'os.exec', 'os.spawn')
    require(not event.startswith(forbidden), f'Replay attempted forbidden action: {event}')
    require(event not in {'os.remove', 'os.rmdir', 'os.rename', 'os.mkdir', 'os.link', 'os.symlink', 'os.truncate', 'os.chmod', 'os.chown', 'os.utime', 'os.system', 'os.fork', 'os.chdir'}, f'Replay attempted mutation: {event}')


def current_candidates(name, workspace):
    if len(name) > 3 and name[:2].isdigit() and name[2] == '-':
        return [workspace / 'project_sources' / name]
    special = {
        'File51a.md': ['project_sources/39-File_51a.Rounded_Scaffold_Mod5_Architecture.md'],
        'File52c_latest.md': ['upload/File_52c.Rounded_Whole_Span_Inverse_Detailed_Study_Draft (2)(1).md'],
        'Strategy.md': ['upload/490d_Unification_Research_Strategy_v0_2_20260906.md', 'c482_c531/evidence/sources/Strategy.md'],
        'C531_checkpoint.md': ['490d_Research_Checkpoint_After_C531_20260927.md'],
        'C531_synthesis.md': ['490d_How_Chronological_Families_Fit_Together_Draft_20260927.md'],
    }
    return [workspace / p for p in special.get(name, [])]


def verify_sources(workspace, require_current):
    manifest = json.loads((ROOT / 'evidence/SOURCE_MANIFEST.json').read_text())
    require(isinstance(manifest, dict) and manifest, 'Missing source manifest')
    require(manifest.get('File52c_latest.md') == PRIMARY_HASH, 'Latest File52c identity mismatch')
    current, absent = [], []
    for name, expected in manifest.items():
        require(Path(name).name == name, f'Unsafe source snapshot path: {name}')
        snapshot = ROOT / 'evidence/sources' / name
        require(digest(snapshot) == expected, f'Source snapshot hash mismatch: {name}')
        available = [p for p in current_candidates(name, workspace) if p.is_file()]
        if not available:
            absent.append(name)
        for path in available:
            require(digest(path) == expected, f'Current source differs from frozen snapshot: {path}')
            current.append({'snapshot': name, 'current_path': str(path), 'sha256': expected})
    require(not require_current or not absent, f'Current originals unavailable: {absent}')
    predecessor = json.loads((ROOT / 'evidence/C531_PREDECESSOR.json').read_text())
    require(predecessor['step'] == 531, 'Wrong predecessor step')
    require(record_digest(predecessor) == predecessor['sha256'] == PREDECESSOR_HASH, 'C531 predecessor hash mismatch')
    require(predecessor['results']['draft_sha256'] == manifest['C531_synthesis.md'], 'C531 synthesis binding mismatch')
    require(predecessor['results']['checkpoint_sha256'] == manifest['C531_checkpoint.md'], 'C531 checkpoint binding mismatch')
    old_journal = workspace / 'c482_c531/journal.json'
    predecessor_live = False
    if old_journal.is_file():
        old_rows = json.loads(old_journal.read_text())
        prior = [r for r in old_rows if r.get('step') == 531]
        require(prior == [predecessor], 'Current C531 record differs from predecessor snapshot')
        predecessor_live = True
    return manifest, current, absent, predecessor_live


def replay(row):
    global REPLAY_ACTIVE
    number = row['step']
    script = ROOT / 'evidence' / f's{number}.py'
    tree = ast.parse(script.read_text(), filename=str(script))
    # An additional static check complements runtime inverse-value tracking.
    for node in ast.walk(tree):
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == 'inv':
            require(not any(isinstance(inner, ast.Call) and isinstance(inner.func, ast.Name) and inner.func.id == 'inv' for arg in node.args for inner in ast.walk(arg)), f'C{number}: nested inverse execution')
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            require(node.name not in {'inv', 'inverse'}, f'C{number}: replay inverse guard was redefined')
    calls = {'begin': 0, 'finish': 0, 'inverse': 0}

    def inverse(n):
        require(isinstance(n, int) and not isinstance(n, bool) and n >= 0, f'C{number}: inverse input must be a nonnegative integer')
        require(not isinstance(n, InverseValue), f'C{number}: forbidden inverse of an inverse-derived value')
        calls['inverse'] += 1
        if n == 0:
            return InverseValue(0)
        # Independent implementation: string-based core extraction, not the production loop.
        text = str(n)
        core = text.rstrip('0')
        return InverseValue(int(core[::-1]) * 10 ** (len(text) - len(core)))

    def begin(n, title, question, inputs, sources):
        calls['begin'] += 1
        require(calls['begin'] == 1 and calls['finish'] == 0, f'C{number}: invalid begin sequence')
        supplied = (n, title, question, clean(inputs), sources)
        recorded = tuple(row[k] for k in ('step', 'title', 'question', 'inputs', 'sources'))
        require(supplied == recorded, f'C{number}: input/source metadata mismatch')
        return copy.deepcopy({k: row[k] for k in ('step', 'title', 'question', 'inputs', 'sources', 'opened_utc', 'predecessor_sha256')})

    def finish(item, results, finding, next_question, checks):
        calls['finish'] += 1
        require(calls['begin'] == calls['finish'] == 1, f'C{number}: invalid finish sequence')
        for key in ('step', 'title', 'question', 'inputs', 'sources', 'opened_utc', 'predecessor_sha256'):
            require(item.get(key) == row[key], f'C{number}: begin record mutated: {key}')
        require(isinstance(checks, dict) and checks and all(v is True for v in checks.values()), f'C{number}: checks are not all Boolean True')
        require((clean(results), finding, next_question, checks) == (row['results'], row['finding'], row['reassessment'], row['checks']), f'C{number}: result/finding/reassessment/check mismatch')

    def artifact(relative_path, content):
        # Recompute each generation event even when a later step revised the file.
        # The historical generated bytes are bound by the result record's digest.
        path = Path(relative_path)
        require(not path.is_absolute() and '..' not in path.parts, f'C{number}: unsafe artifact path')
        data = content.encode()
        return {'path': relative_path, 'sha256': hashlib.sha256(data).hexdigest(), 'bytes': len(data)}

    fake = types.ModuleType('research')
    exported = {'Path': Path, 'F': Fraction, 'ROOT': ROOT, 'E': Fraction(25, 23), 'P': Fraction(70, 69), 'J': Fraction(300, 299), 'json': json, 'hashlib': hashlib, 'clean': clean, 'inv': inverse, 'begin': begin, 'finish': finish, 'REPLAY': True, 'artifact': artifact}
    for key, value in exported.items():
        setattr(fake, key, value)
    previous = sys.modules.get('research')
    sys.modules['research'] = fake
    capture = io.StringIO()
    try:
        REPLAY_ACTIVE = True
        with contextlib.redirect_stdout(capture):
            runpy.run_path(str(script), run_name=f'replay_{number}')
    finally:
        REPLAY_ACTIVE = False
        if previous is None:
            sys.modules.pop('research', None)
        else:
            sys.modules['research'] = previous
    require(calls['begin'] == calls['finish'] == 1, f'C{number}: missing begin or finish')
    return {'step': number, 'script_sha256': digest(script), 'checks': len(row['checks']), 'single_pass_inverse_calls': calls['inverse']}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--through', type=int, default=LAST, help='Completed last step to replay; default631 requires all100')
    parser.add_argument('--workspace', type=Path, default=ROOT.parent, help='Optional workspace holding current originals; default packet parent')
    parser.add_argument('--require-current-sources', action='store_true', help='Fail if any current original is absent; omit for extracted portable packets')
    parser.add_argument('--output', type=Path, help='Optional report path; the only file written by this runner')
    args = parser.parse_args()
    require(FIRST <= args.through <= LAST, f'--through must be in {FIRST}..{LAST}')
    raw = (ROOT / 'journal.json').read_bytes()
    rows = json.loads(raw)
    require(isinstance(rows, list) and rows, 'Empty journal')
    require(len(rows) <= LAST - FIRST + 1, 'Journal exceeds requested100-step range')
    require([r['step'] for r in rows] == list(range(FIRST, FIRST + len(rows))), 'Nonsequential/duplicate journal steps')
    require(rows[-1]['step'] >= args.through, f'Requested C{args.through} is unfinished; use --through {rows[-1]["step"]}')
    selected = [r for r in rows if r['step'] <= args.through]
    if args.through == LAST:
        require(len(rows) == 100, 'Final verification requires all100 completed steps')
        require(not (ROOT / 'pending.json').exists(), 'Final verification found an unfinished pending step')
    required = {'step', 'title', 'question', 'inputs', 'sources', 'opened_utc', 'predecessor_sha256', 'results', 'finding', 'reassessment', 'checks', 'closed_utc', 'sha256'}
    expected_previous = 'C531:' + PREDECESSOR_HASH
    previous_close = None
    for row in selected:
        require(set(row) == required, f'C{row["step"]}: unexpected record schema')
        require(record_digest(row) == row['sha256'], f'C{row["step"]}: record hash mismatch')
        require(row['predecessor_sha256'] == expected_previous, f'C{row["step"]}: predecessor chain mismatch')
        require(all(isinstance(row[k], str) and row[k].strip() for k in ('title', 'question', 'finding', 'reassessment')), f'C{row["step"]}: empty research metadata')
        require(isinstance(row['sources'], list) and row['sources'], f'C{row["step"]}: missing sources')
        opened, closed = datetime.fromisoformat(row['opened_utc']), datetime.fromisoformat(row['closed_utc'])
        require(opened.tzinfo is not None and closed.tzinfo is not None and opened <= closed, f'C{row["step"]}: invalid timestamps')
        require(previous_close is None or previous_close <= opened, f'C{row["step"]}: overlapping numbered sequence')
        previous_close, expected_previous = closed, row['sha256']
    manifest, current, absent, predecessor_live = verify_sources(args.workspace.resolve(), args.require_current_sources)
    sys.dont_write_bytecode = True
    sys.addaudithook(audit_read_only)
    replays = [replay(row) for row in selected]
    report = {
        'status': 'PASS', 'first': FIRST, 'last': args.through, 'sequential_steps': len(selected),
        'final_100_step_verification': args.through == LAST, 'journal_steps_available_at_read': len(rows),
        'journal_sha256_at_read': hashlib.sha256(raw).hexdigest(), 'last_record_sha256': expected_previous,
        'record_metadata_and_results_replayed': True, 'journal_hash_chain_verified': True,
        'C531_predecessor_snapshot_verified': True, 'C531_current_record_checked': predecessor_live,
        'source_snapshot_hashes_verified': len(manifest), 'current_source_copies_checked': len(current),
        'current_sources': current, 'current_originals_absent': absent,
        'latest_File52c_sha256': PRIMARY_HASH,
        'step_checks_replayed': sum(r['checks'] for r in replays),
        'independent_reversal_used': True, 'single_pass_inverse_calls': sum(r['single_pass_inverse_calls'] for r in replays),
        'second_inverse_guard_violations': 0,
        'inverse_guard_scope': 'Nested inv calls and inversion of provenance-tracked inv outputs are rejected. This is not a proof against arbitrary manual reimplementation or deliberate provenance erasure.',
        'replay_writes_blocked': True, 'replays': replays,
        'independent_review_notes': sorted(p.name for p in (ROOT / 'evidence').glob('REVIEW_*.md')),
        'scope': 'Exact record and script replay, source identity, and declared operation checks. Check counts are not independent discoveries or statistical evidence. Historical source suites were not rerun.'
    }
    text = json.dumps(report, indent=2, ensure_ascii=False) + '\n'
    if args.output:
        target = args.output.resolve()
        require(target != ROOT / 'journal.json' and target.suffix == '.json', 'Output must be a separate JSON report')
        target.write_text(text)
    print(text, end='')


if __name__ == '__main__':
    try:
        main()
    except (VerificationError, KeyError, ValueError, OSError) as exc:
        print(json.dumps({'status': 'FAIL', 'error': str(exc)}, ensure_ascii=False), file=sys.stderr)
        sys.exit(1)